#!/usr/bin/env python3
"""nextest 计数锚换算器 (v0.98.3 M1 S2, openspec/changes/v0983-test-infra/design.md §3.2)

锚口径 (ar 裁定 09-02): nextest 计数 + `cargo test --doc` doctest 计数合并
  —— nextest 不跑 doctest, 锚三元组 (passed/failed/ignored) 与 doctest 计数分列。

对账纪律 (claude-test O-1 / ar 00:23 硬要求):
  对账单位 = 测试名多重集合 (Counter), 禁用聚合计数互推——nextest 汇总行的
  skipped 混合"过滤不匹配"与"真 ignore", 与 cargo 侧 ignored 字段口径不同源,
  汇总对汇总必错。diff 逐测试名报告。

用法:
  --self-test          构造样本流自测解析与对账逻辑 (CI/门禁可调用)
  --from-files N.json C.txt D.txt
                       从落盘输出解析 (nextest libtest-json / cargo test 输出 /
                       cargo test --doc 输出), 供复算与留痕
  --compare-vs-cargo   本机实测: 依次跑 nextest 全量(json) + cargo test --tests
                       + cargo test --doc, 解析并集合级对账, 不一致即退出码 1
                       (O-2: workspace 全量执行, 编译段墙钟与执行段时间分开报告)
"""
from __future__ import annotations

import argparse
import collections
import json
import re
import subprocess
import sys
import time
from dataclasses import dataclass, field

# ── 解析器 ──────────────────────────────────────────────────────────

# nextest libtest-json 事件: name 形如 "xw-tui::xw_tui$app::tests::r19_x"
# 规范化 = 取 $ 后的测试路径 (与 cargo test 输出的 test NAME 对齐;
# 多 binary 同名测试由 Counter 计数对齐)。
_NEXTTEST_NAME = re.compile(r"\$(.+)$")


def _norm_nextest_name(name: str) -> str:
    m = _NEXTTEST_NAME.search(name)
    return m.group(1) if m else name


def parse_nextest_json(stream: str) -> tuple[collections.Counter, dict]:
    """解析 nextest --message-format libtest-json 事件流。

    返回 (Counter{test_path: count}, meta)；meta 含 exec_time 总和与
    top 10 最耗时测试 (W1 遗留: 单测时长实证供数)。
    event 取值: started/ok/failed/ignored (skipped= 过滤不匹配, 不计入锚)。
    """
    counter: collections.Counter = collections.Counter()
    failed: collections.Counter = collections.Counter()
    ignored: collections.Counter = collections.Counter()
    started: set[str] = set()
    exec_times: list[tuple[float, str]] = []
    non_json_lines = 0  # 捕获卫生: >0 = json 流混入人类输出（stdout+stderr 合并等）
    for line in stream.splitlines():
        line = line.strip()
        if not line or not line.startswith("{"):
            if line:
                non_json_lines += 1
            continue
        try:
            e = json.loads(line)
        except json.JSONDecodeError:
            non_json_lines += 1
            continue
        if e.get("type") != "test":
            continue
        name = _norm_nextest_name(e.get("name", ""))
        ev = e.get("event")
        if ev == "started":
            started.add(name)
        elif ev == "ok":
            counter[name] += 1
            if "exec_time" in e:
                exec_times.append((float(e["exec_time"]), name))
        elif ev == "failed":
            failed[name] += 1
        elif ev == "ignored":
            ignored[name] += 1
    exec_times.sort(reverse=True)
    finals = counter + failed + ignored
    missing_final = started - set(finals.elements())
    meta = {
        "passed": sum(counter.values()),
        "failed": sum(failed.values()),
        "ignored": sum(ignored.values()),
        "started_total": len(started),
        "final_total": sum(finals.values()),
        "missing_final": sorted(missing_final),
        "ignored_counter": dict(ignored),
        "non_json_lines": non_json_lines,
        "exec_total": round(sum(t for t, _ in exec_times), 3),
        "top_slow": [(round(t, 3), n) for t, n in exec_times[:10]],
    }
    return counter, meta


_NEXTTEST_FAILED = re.compile(r'\{"type":"test","event":"failed","name":"([^"]+)"')


def _nextest_failed_names(stream: str) -> list[str]:
    """红测名单 (规范化名, 排序去重)——对账报告点名用。"""
    return sorted({_norm_nextest_name(m) for m in _NEXTTEST_FAILED.findall(stream)})


# OSC/CSI 转义剥离: TUI 测试进程的终端查询序列 (如 \x1b]11;?\x1b\\ 背景色探测)
# 会粘连在 "test NAME ... ok" 行前, 破坏行首锚定 (S2 首对账 only_a 10 项假差异根因)
def _strip_ansi(s: str) -> str:
    """剥离 ANSI 转义序列 (手写状态机, 零正则转义歧义)。

    背景 (S2 首对账实证): TUI 测试进程的终端查询序列 (OSC 11 背景色探测
    ESC ]11;? ESC \) 粘连在 "test NAME ... ok" 行前, 破坏行首锚定 →
    only_a 假差异 10 项。覆盖: CSI (ESC[...最终字节@-~) / OSC (...BEL 或
    ST=ESC\) / 其他 Fe 单字节序列。
    """
    ESC, BEL = chr(27), chr(7)
    out: list[str] = []
    i, n = 0, len(s)
    while i < n:
        c = s[i]
        if c != ESC:
            out.append(c)
            i += 1
            continue
        if i + 1 < n and s[i + 1] == "[":
            j = i + 2
            while j < n and not ("@" <= s[j] <= "~"):
                j += 1
            i = j + 1
        elif i + 1 < n and s[i + 1] == "]":
            j = i + 2
            while j < n:
                if s[j] == BEL:
                    j += 1
                    break
                if s[j] == ESC and j + 1 < n and s[j + 1] == chr(92):
                    j += 2
                    break
                j += 1
            i = j
        else:
            i += 2
    return "".join(out)


_CARGO_TEST_LINE = re.compile(r"^test (\S+?)(?: - should panic)? \.\.\. (ok|FAILED|ignored)")


def parse_cargo_test(stream: str) -> tuple[collections.Counter, collections.Counter, collections.Counter]:
    """解析 cargo test (human 格式) 输出 → (passed, failed, ignored) 多重集合。

    多 target 同名测试自然累加计数 (multiset 语义, 与 nextest 侧规范化后对齐)。
    """
    passed: collections.Counter = collections.Counter()
    failed: collections.Counter = collections.Counter()
    ignored: collections.Counter = collections.Counter()
    for line in _strip_ansi(stream).splitlines():
        m = _CARGO_TEST_LINE.match(line.strip())
        if not m:
            continue
        name, status = m.group(1), m.group(2)
        (passed if status == "ok" else failed if status == "FAILED" else ignored)[name] += 1
    return passed, failed, ignored


_CARGO_DOC_LINE = re.compile(r"^test (\S+ - .+?) \.\.\. (ok|FAILED|ignored)")


def parse_cargo_doctest(stream: str) -> tuple[collections.Counter, collections.Counter, collections.Counter]:
    """解析 cargo test --doc 输出 → (passed, failed, ignored)。

    doctest 名形如 "src/lib.rs - ident (line N)"，用独立匹配器避免与
    普通测试行混淆 (doctest 名含 " - ", 普通测试名不含)。
    """
    passed: collections.Counter = collections.Counter()
    failed: collections.Counter = collections.Counter()
    ignored: collections.Counter = collections.Counter()
    for line in _strip_ansi(stream).splitlines():
        m = _CARGO_DOC_LINE.match(line.strip())
        if not m:
            continue
        name, status = m.group(1), m.group(2)
        (passed if status == "ok" else failed if status == "FAILED" else ignored)[name] += 1
    return passed, failed, ignored


# ── 对账 (O-1: 测试名级) ───────────────────────────────────────────


def diff_counters(a: dict, b: dict) -> dict:
    """多重集合 diff: a 有 b 无 / b 有 a 无 / 两侧都有但计数不等。

    宽容 dict 入参 (内部转 Counter); self-test 直接传 dict 字面量。
    """
    a, b = collections.Counter(a), collections.Counter(b)
    # 对账语义: 单侧完全缺失进 only_*; 两侧都有但计数不等进 count_mismatch
    # (Counter 的 -/& 运算保留 min/正余量, 不符合"计数差"对账语义, 故手工分桶)
    only_a = {n: a[n] for n in a if n not in b}
    only_b = {n: b[n] for n in b if n not in a}
    count_mismatch = {
        n: {"a": a[n], "b": b[n]} for n in a.keys() & b.keys() if a[n] != b[n]
    }
    return {
        "equal": not (only_a or only_b or count_mismatch),
        "only_a": dict(only_a),
        "only_b": dict(only_b),
        "count_mismatch": count_mismatch,
    }


def reconcile(
    n_counter: collections.Counter,
    n_meta: dict,
    c_pass: collections.Counter,
    c_ign: collections.Counter,
    d_pass: collections.Counter,
    d_fail: collections.Counter,
    d_ign: collections.Counter,
    human_log_hint: str = "",
) -> tuple[int, list[str]]:
    """共享对账核心 (claude-test 05:35 修复规格): compare_vs_cargo 与 from_files
    两路径必须同走同一套断言——证据重放与现场对账等价。

    断言序 (ar 采信规格):
      1. 捕获卫生: nextest JSON 流混入非 JSON 行 → FAIL (旧缺陷根因)
      2. 完整性精化: missing_final 与 cargo ignored 集比对——全部可解释
         (静态 #[ignore]) → WARN+以 cargo 口径为准; 不可解释缺名 → FAIL
      3. passed/ignored 名级 multiset 双向对账——passed 差异 FAIL;
         ignored 差异若全部由 missing_final 解释 → WARN(cargo 口径),
         否则 FAIL(交叉不一致)

    返回 (exit_rc, report_lines)。rc: 0=一致 1=对账差异 2=断言 FAIL。
    """
    lines: list[str] = []
    rc = 0

    # 1) 捕获卫生
    if n_meta.get("non_json_lines", 0) > 0:
        rc = 2
        lines.append(
            f"[FAIL] 捕获卫生: nextest JSON 流混入 {n_meta['non_json_lines']} 条"
            f"非 JSON 行{human_log_hint}——stdout 与 stderr 必须分离落盘"
        )

    # 2) 完整性精化: missing_final vs cargo ignored 集
    missing = set(n_meta.get("missing_final", []))
    if missing:
        explicable = missing & set(c_ign)
        unexplained = missing - set(c_ign)
        if unexplained:
            rc = max(rc, 2)
            lines.append(
                f"[FAIL] 完整性: {len(unexplained)} 个 started 无 final 且不在 "
                f"cargo ignored 集（不可解释）: {sorted(unexplained)[:10]}"
            )
        else:
            lines.append(
                f"[WARN] 完整性: {len(missing)} 个静态 #[ignore] 缺 final 事件"
                f"（全部在 cargo ignored 集可解释）——ignored 锚以 cargo 口径为准"
            )

    # 3a) passed multiset 对账
    d1 = diff_counters(n_counter, c_pass)
    if not d1["equal"]:
        rc = max(rc, 1)
    lines += report_diff("passed 集合 (nextest vs cargo --tests, multiset)", d1)

    # 3b) ignored 名级对账: 差异须全部由 missing_final 解释
    n_ign = collections.Counter(n_meta.get("ignored_counter", {}))
    d2 = diff_counters(n_ign, c_ign)
    only_b_names = set(d2["only_b"])
    only_a_names = set(d2["only_a"])
    covered_by_missing = only_b_names <= missing
    unexplained_ign = only_a_names or (only_b_names - missing) or d2["count_mismatch"]
    if not d2["equal"]:
        if covered_by_missing and not unexplained_ign:
            lines.append(
                f"[WARN] ignored 名级差异 {len(only_b_names)} 项全部由 "
                f"missing_final 解释——cargo 口径采信"
            )
        else:
            rc = max(rc, 1)
            lines += report_diff("ignored 集合 (nextest vs cargo, multiset)", d2)
            lines.append("[FAIL] ignored 交叉不一致且无法由 missing_final 解释")

    # 锚合并
    total_p = n_meta["passed"] + sum(d_pass.values())
    total_f = n_meta["failed"] + sum(d_fail.values())
    total_i = sum(c_ign.values()) + sum(d_ign.values())
    lines.append(
        f"锚合并: {total_p} passed / {total_f} failed / {total_i} ignored "
        f"(ignored=cargo 人类口径 {sum(c_ign.values())}+doc {sum(d_ign.values())})"
    )
    return rc, lines


def report_diff(label: str, d: dict, max_show: int = 20) -> list[str]:
    lines = [f"[{label}] {'一致 ✓' if d['equal'] else '存在差异 ✗'}"]
    for key in ("only_a", "only_b"):
        items = list(d[key].items())[:max_show]
        lines.append(f"  {key} ({len(d[key])}): {[n for n, _ in items]}")
        if len(d[key]) > max_show:
            lines.append(f"    ... 其余 {len(d[key]) - max_show} 项省略")
    for n, c in list(d["count_mismatch"].items())[:max_show]:
        lines.append(f"  count_mismatch: {n} a={c['a']} b={c['b']}")
    return lines


# ── 实跑对账 (O-2: workspace 全量) ──────────────────────────────────

ENV_STRIP = ("XWCODE_ALLOW_DANGEROUS",)  # S3 教训: 外泄使 block 测试全挂


def _run(cmd: list[str], env: dict) -> tuple[str, str, float]:
    """执行命令, 分离返回 (stdout, stderr, 墙钟)。

    捕获卫生 (claude-test AC1 复算 03:21): json 流与人类输出禁同文件——
    stdout(JSON) 单独落盘, stderr(人类 SKIP/进度行)另行记录, 合并落盘会
    混入非 JSON 行且掩盖 ignored 事件缺失。
    """
    t0 = time.monotonic()
    r = subprocess.run(cmd, capture_output=True, text=True, env=env)
    wall = time.monotonic() - t0
    return r.stdout, r.stderr, wall


def compare_vs_cargo(save_dir: str | None = None) -> int:
    """三跑实测对账: nextest 全量 / cargo test --tests 全量 / cargo test --doc 全量。

    cargo 侧 --no-fail-fast: 对账要求双侧全量跑完 (fail-fast 中止后续 binary
    会使 cargo passed 集合残缺, 首跑实证: 1 红测中止 → 12 项假性缺失)。
    save_dir 提供时落盘三份原始输出 (留痕可复算)。
    """
    env = dict(sys_exec_env())
    for k in ENV_STRIP:
        env.pop(k, None)
    env.setdefault("NEXTEST_EXPERIMENTAL_LIBTEST_JSON", "1")

    print("== [1/3] cargo nextest run --workspace (libtest-json) ==", flush=True)
    n_out, n_err, n_wall = _run(
        ["cargo", "nextest", "run", "--workspace", "--message-format", "libtest-json"], env
    )
    n_counter, n_meta = parse_nextest_json(n_out)

    print("== [2/3] cargo test --workspace --tests --no-fail-fast ==", flush=True)
    c_out, c_err, c_wall = _run(["cargo", "test", "--workspace", "--tests", "--no-fail-fast"], env)
    c_pass, c_fail, c_ign = parse_cargo_test(c_out)

    print("== [3/3] cargo test --workspace --doc ==", flush=True)
    d_out, d_err, d_wall = _run(["cargo", "test", "--workspace", "--doc"], env)
    d_pass, d_fail, d_ign = parse_cargo_doctest(d_out)

    if save_dir:
        import pathlib
        p = pathlib.Path(save_dir)
        p.mkdir(parents=True, exist_ok=True)
        (p / "nextest.json").write_text(n_out)       # 仅 JSON 流 (捕获卫生)
        (p / "nextest.human.log").write_text(n_err)  # 人类输出分离记录
        (p / "cargo_tests.txt").write_text(c_out)
        (p / "cargo_doc.txt").write_text(d_out)
        print(f"\n原始输出已落盘 {p}/ (复算: --from-files ...; json 与人类输出已分离)")

    # 共享对账 (claude-test 05:35 修复规格#1: 两路径同走同一套断言)
    rc, report = reconcile(n_counter, n_meta, c_pass, c_ign, d_pass, d_fail, d_ign)
    print()
    for line in report:
        print(line)
    if n_meta["failed"] > 0:
        failed_names = _nextest_failed_names(n_out)
        if failed_names:
            print(f"  nextest 红测名单 ({len(failed_names)}): {failed_names}")
    return rc

    print(f"\nnextest 墙钟 {n_wall:.1f}s | cargo --tests 墙钟 {c_wall:.1f}s | "
          f"cargo --doc 墙钟 {d_wall:.1f}s (含编译段)")
    print(f"nextest 执行段(测试内)合计 {n_meta['exec_total']}s | "
          f"top slow: {n_meta['top_slow'][:5]}")

    print(f"\nnextest: passed={n_meta['passed']} failed={n_meta['failed']} "
          f"ignored={n_meta['ignored']} | cargo --tests 参照: "
          f"{sum(c_pass.values())}/{sum(c_fail.values())}/{sum(c_ign.values())}")
    return rc


def sys_exec_env() -> dict:
    return dict(__import__("os").environ)


# ── 自测 ───────────────────────────────────────────────────────────


def self_test() -> int:
    """构造样本流自测: 解析正确性 + 规范化 + multiset 对账 (ar 验收要求)。"""
    checks: list[tuple[str, bool]] = []

    # 1) nextest json: ok/failed/ignored/skipped 语义 + name 规范化
    nxt = "\n".join([
        '{"type":"suite","event":"started","test_count":4}',
        '{"type":"test","event":"started","name":"xw-tui::xw_tui$app::tests::a"}',
        '{"type":"test","event":"ok","name":"xw-tui::xw_tui$app::tests::a","exec_time":0.01}',
        '{"type":"test","event":"failed","name":"xw-tui::types_v0700$test_b"}',
        '{"type":"test","event":"ignored","name":"xw-tui::xw_tui$app::tests::c"}',
        '{"type":"test","event":"started","name":"xw-tui::xw_tui$app::tests::skipme"}',
        '{"type":"test","event":"ok","name":"xw-tui::xw_tui$app::tests::skipme","exec_time":0.02}',
        'non-json noise line',
    ])
    c1, m1 = parse_nextest_json(nxt)
    checks.append(("nextest: 规范化去 $ 前缀", c1 == {"app::tests::a": 1, "app::tests::skipme": 1}))
    checks.append(("nextest: ok/failed/ignored 分列", m1["passed"] == 2 and m1["failed"] == 1 and m1["ignored"] == 1))
    checks.append(("nextest: exec_total 与 top_slow", abs(m1["exec_total"] - 0.03) < 1e-9 and m1["top_slow"][0][1] == "app::tests::skipme"))

    # 2) cargo test 行: ok/FAILED/ignored + 多 target 同名累加
    cg = "\n".join([
        "test app::tests::a ... ok",
        "test app::tests::a ... ok",          # 另一 target 同名 → 计数 2
        "test test_b ... FAILED",
        "test app::tests::c ... ignored",
        "test test_panic_case - should panic ... ok",   # should_panic 形态 (首跑实证漏匹配)
        chr(27) + "]11;?" + chr(27) + chr(92) + "test app::tests::osc_glued ... ok",  # OSC 粘连 (二跑实证漏匹配)
        "running 10 tests",
        "test result: ok. 4 passed; 0 failed",
    ])
    p2, f2, i2 = parse_cargo_test(cg)
    checks.append(("cargo: multiset 计数(同名跨target=2)+should_panic 形态",
                   p2 == {"app::tests::a": 2, "test_panic_case": 1, "app::tests::osc_glued": 1}
                   and f2 == {"test_b": 1} and i2 == {"app::tests::c": 1}))

    # 3) doctest 行独立匹配 (不与普通测试行混淆)
    dc = "\n".join([
        "test src/lib.rs - hello (line 5) ... ok",
        "test src/tools.rs - example (line 12) ... ok",
        "test app::tests::a ... ok",          # 普通行不应被 doctest 匹配
        "test src/lib.rs - bad (line 9) ... FAILED",
    ])
    p3, f3, _ = parse_cargo_doctest(dc)
    checks.append(("doctest: 仅匹配 doc 样式名", p3 == {"src/lib.rs - hello (line 5)": 1, "src/tools.rs - example (line 12)": 1} and f3 == {"src/lib.rs - bad (line 9)": 1}))

    # 4) multiset diff: 相等/缺失/多余/计数差
    d_eq = diff_counters({"a": 2, "b": 1}, {"a": 2, "b": 1})
    d_ne = diff_counters({"a": 2}, {"a": 1, "b": 3})
    checks.append(("diff: 相等集合", d_eq["equal"]))
    checks.append(("diff: 计数差+缺失+多余检出",
                   not d_ne["equal"] and d_ne["only_a"] == {} and d_ne["only_b"] == {"b": 3}
                   and d_ne["count_mismatch"] == {"a": {"a": 2, "b": 1}}))

    # 5) 锚合并口径: nextest + doctest
    merged = (m1["passed"] + sum(p3.values()), m1["failed"] + sum(f3.values()), m1["ignored"])
    checks.append(("锚合并: nextest+doctest 计数", merged == (4, 2, 1)))

    ok = True
    for name, passed in checks:
        print(f"  {'PASS' if passed else 'FAIL'}  {name}")
        ok &= passed
    print(f"self-test: {'ALL PASS' if ok else 'FAILED'} ({len(checks)} checks)")
    return 0 if ok else 2


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--self-test", action="store_true", help="构造样本流自测")
    ap.add_argument("--from-files", nargs=3, metavar=("NEXT_JSON", "CARGO_TXT", "DOC_TXT"),
                    help="从落盘输出解析并对账")
    ap.add_argument("--compare-vs-cargo", action="store_true",
                    help="实跑三命令全量对账 (O-2 workspace 全量)")
    ap.add_argument("--save-dir", metavar="DIR",
                    help="compare-vs-cargo 模式下落盘三份原始输出 (留痕可复算)")
    args = ap.parse_args()
    if args.self_test:
        return self_test()
    if args.from_files:
        # v0.98.3 M3 修正#2 (claude-test 05:35): from_files 与 live 同走共享
        # reconcile——证据重放与现场对账等价（旧版本分支零断言, 旧缺陷捕获
        # 喂入仍 exit0 报'一致'——AC1 闭环被阻塞的根因）。
        n_counter, n_meta = parse_nextest_json(open(args.from_files[0]).read())
        c_pass, c_fail, c_ign = parse_cargo_test(open(args.from_files[1]).read())
        d_pass, d_fail, d_ign = parse_cargo_doctest(open(args.from_files[2]).read())
        rc, report = reconcile(n_counter, n_meta, c_pass, c_ign, d_pass, d_fail, d_ign)
        print("\n".join(report))
        return rc
    if args.compare_vs_cargo:
        return compare_vs_cargo(save_dir=args.save_dir)
    ap.print_help()
    return 0


if __name__ == "__main__":
    sys.exit(main())