"""托管 shell 脚本：管道任一读取级不得是提前退出的读取器（READER_FAMILIES 声明的各族：grep -q/-m、head、sed q、
awk exit/nextfile、read、mapfile/readarray -n、以 break/exit 结束或 until 条件含读取器的 while/until/select 循环；
含被子壳或花括组包裹、带绝对路径或 env/exec/nohup/timeout 等前缀的写法）。守护的承诺是“已声明的各族在发现到的
脚本中处处被拦截”，而非“理解任意 bash 命令”。

pipefail 下读取器先退出，上游生产者写满 64 KiB 管道缓冲后收到 SIGPIPE，管道状态报 141，
`if producer | grep -q PAT` 与 `if ! ...` 的判定被静默翻转；读取器位于管道中段
（`producer | head -1 | sed ...`）时同理。改用 here-string、进程替换或 sed 打印即退出，
读取器的结论才不会丢。
"""
import re
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SCRIPT_DIRS = ("scripts", "docker")
EXTRA_SCRIPTS = ("docker/verify-env.sh", "tests/load/collaboration/run_pgbench.sh")
SHELL_SHEBANG = re.compile(r"#!.*\b(?:ba|da|k|z)?sh\b")
# 发现到但检测器无法正确解析的脚本：路径 -> 理由。条目须仍被发现，否则视为过期。
EXEMPT = {}

READER_FAMILIES = (
    ("grep", "-q/--quiet/--silent、-m/--max-count（含 egrep/fgrep）", "得出匹配结论或读满 N 条匹配即退出"),
    ("head", "任意用法", "输出前 N 行/字节后即退出"),
    ("sed", "q/Q 命令", "执行 q/Q 即停止读取"),
    ("awk", "END 块之外的 exit/nextfile（含 gawk/mawk/nawk）", "exit 立即结束，nextfile 放弃当前输入剩余部分"),
    ("read", "任意用法（含 -n N/--chars N）", "只读一行，-n N 读满 N 个字符即返回"),
    ("mapfile", "-n/-n<N>/-n <N>/--n/--numlines（含 readarray）", "读满 N 行即停止；长选项被 bash 拒绝时一字节不读即退出"),
    (
        "while",
        "while/until/select 循环：循环内命令位的 break 出本循环、命令位的 exit（子壳内除外）、until 条件含上列读取器"
        "（! 取反除外）",
        "break/exit 结束循环后不再读取剩余输入；until 条件读取成功即结束循环",
    ),
)
"""守护覆盖的提前退出读取器族：(命令, 触发选项, 提前退出的原因)。

检测器保证这些族（含子壳/花括组包裹、VAR= 前缀、!/time/command/builtin/env/exec 前缀、nohup/timeout/nice/stdbuf
包装命令及其选项与 timeout 时长、绝对路径写法）位于管道读取级时处处被拦截；不带触发选项的 mapfile/readarray、
无 break/exit 的 while read、条件含读取器的 while（条件失败即 EOF 才结束）读完全部输入，不在拦截之列。
"""

HEREDOC = re.compile(r"<<(-?)\s*(['\"]?)([A-Za-z_][A-Za-z0-9_]*)\2")
SED_QUIT = re.compile(r"(?:^|[\s;{}'\"/$0-9])[qQ][0-9]*\s*(?:$|[\s;}'\"])")
AWK_END = re.compile(r"\bEND\s*\{")
AWK_EXIT = re.compile(r"\b(?:exit|nextfile)\b")
AWK_STRING = re.compile(r'"(?:[^"\\\n]|\\.)*"')
SHELL_SQ = re.compile(r"'[^']*'")
ASSIGNMENT = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=")
WORD = re.compile(r"(\S+)\s*")
BREAK_ARG = re.compile(r"[ \t]+([0-9]+)")
PREFIXES = {"!", "time", "command", "builtin", "env", "exec"}
# 包装命令：(带独立参数的选项, 被包装命令前的操作数个数)；均在首个操作数处停止选项解析。
WRAPPERS = {
    "nohup": (set(), 0),
    "timeout": ({"-s", "-k", "--signal", "--kill-after"}, 1),
    "nice": ({"-n", "--adjustment"}, 0),
    "stdbuf": ({"-i", "-o", "-e", "--input", "--output", "--error"}, 0),
}
LOOPS = ("while", "until", "select")
GREP_ARG_OPTS = set("efABCdDX")
MAPFILE_ARG_OPTS = set("dOsuCc")
BOUNDARY = set(" \t\n;&|()<>")


def _is_shell_script(rel):
    path = ROOT / rel
    if rel.endswith(".sh"):
        return True
    if not path.is_file():
        return False
    with path.open(encoding="utf-8", errors="replace") as f:
        return bool(SHELL_SHEBANG.match(f.readline()))


def managed_scripts():
    """git 跟踪的托管 shell 脚本（.sh 或首行 shell shebang）；未跟踪副本（如 runtime/ 下的陈旧克隆）不在其中。"""
    out = subprocess.run(
        ["git", "-C", str(ROOT), "ls-files", "-z", "--", *SCRIPT_DIRS, *EXTRA_SCRIPTS],
        check=True, capture_output=True, text=True,
    ).stdout
    return sorted(p for p in set(out.split("\0")) if p and _is_shell_script(p))


class _Frame:
    """一层命令上下文（顶层、$(...)、(...)、<(...)、管道级的 { ...; }、while/until/select 循环）内正在累积的管道。"""

    def __init__(self, line, depth, closer=None, kind=None, cond_start=0):
        self.stages, self.stage, self.start, self.cases = [], [], line, []
        self.depth, self.closer, self.dos, self.broken = depth, closer, 0, False
        self.kind, self.cond_start, self.cond = kind, cond_start, []


def _lex(src):
    """词法扫描 bash 源码；跳过注释与 heredoc 正文。

    返回 (lists, loops)：lists 为每个命令列表的 (帧深度, [(起始行号, 级文本), ...])，
    loops 为每个 while/until/select 循环的 (帧深度, 循环关键字, 是否被 break/exit 结束, 条件内的 lists)。
    """
    found, loops, frames, modes, heredocs = [], [], [_Frame(1, 0)], ["cmd"], []
    line, word, i, n = 1, "", 0, len(src)

    def emit(s):
        for f in frames:
            f.stage.append(s)

    def finish(f):
        text = "".join(f.stage).strip()
        if f.stages or text:
            found.append((f.depth, f.stages + [(f.start, text)]))
        f.stages, f.stage, f.start = [], [], line

    def push(closer, kind=None):
        modes.append("cmd")
        frames.append(_Frame(line, len(frames), closer, kind, len(found)))

    def pop():
        finish(frames[-1])
        f = frames.pop()
        modes.pop()
        if f.closer == "done":
            loops.append((f.depth, f.kind, f.broken, f.cond))

    def case_state():
        cases = frames[-1].cases
        return cases[-1] if cases else None

    def command_position(w):
        j = i - len(w)
        while j > 0 and src[j - 1] in " \t":
            j -= 1
        if j == 0 or src[j - 1] in ";\n&|(" or (src[j - 1] == ")" and case_state() == "body"):
            return True
        k = j
        while k > 0 and src[k - 1] not in BOUNDARY:
            k -= 1
        return src[k:j] in ("do", "then", "else", "{", "!")

    def break_out(levels):
        for f in reversed(frames):
            nested = f.dos - (f.closer == "done" and f.dos > 0)
            if levels <= nested:
                return
            levels -= nested
            if f.closer == "done":
                f.broken = True
                levels -= 1
                if levels <= 0:
                    return
            elif f.closer != "}":
                return

    def exit_out():
        for f in reversed(frames):
            if f.closer == "done":
                f.broken = True
            elif f.closer != "}":
                return

    def keyword(w):
        top = frames[-1]
        cases = top.cases
        if w == "case":
            cases.append("in")
        elif w == "in" and case_state() == "in":
            cases[-1] = "pattern"
        elif w == "esac" and cases:
            cases.pop()
        elif case_state() == "pattern":
            pass
        elif w in LOOPS and "".join(top.stage).strip() == w:
            push("done", w)
        elif w == "do" and command_position(w):
            if top.closer == "done" and top.dos == 0:
                top.cond = found[top.cond_start :]
            top.dos += 1
        elif w == "done" and command_position(w):
            top.dos -= 1
            if top.closer == "done" and top.dos <= 0:
                pop()
        elif w == "break" and command_position(w):
            m = BREAK_ARG.match(src, i)
            break_out(int(m.group(1)) if m else 1)
        elif w == "exit" and command_position(w):
            exit_out()

    def push_cmd(opener, closer=")"):
        emit(opener)
        push(closer)

    while i < n:
        c, nxt = src[i], src[i + 1 : i + 2]
        mode = modes[-1]
        if mode in ("sq", "ansi"):
            if c == "\\" and mode == "ansi":
                emit(src[i : i + 2])
                i += 2
                continue
            emit(c)
            line += c == "\n"
            if c == "'":
                modes.pop()
            i += 1
            continue
        if mode in ("dq", "brace"):
            if c == "\\":
                emit(src[i : i + 2])
                line += nxt == "\n"
                i += 2
            elif c == "$" and nxt == "(":
                push_cmd("$(")
                i += 2
            elif c == "$" and nxt == "{":
                emit("${")
                modes.append("brace")
                i += 2
            elif c == '"':
                emit(c)
                if mode == "dq":
                    modes.pop()
                else:
                    modes.append("dq")
                i += 1
            elif c == "}" and mode == "brace":
                emit(c)
                modes.pop()
                i += 1
            else:
                emit(c)
                line += c == "\n"
                i += 1
            continue

        if c in BOUNDARY and word:
            keyword(word)
            word = ""
        top = frames[-1]
        if c == "\\":
            if nxt == "\n":
                emit(" ")
                line += 1
            else:
                emit(src[i : i + 2])
                word += "\0"
            i += 2
        elif c == "#" and (i == 0 or src[i - 1] in " \t\n;&|("):
            j = src.find("\n", i)
            i = n if j < 0 else j
        elif c == "'" or (c == "$" and nxt == "'"):
            emit(src[i : i + 2] if c == "$" else c)
            modes.append("ansi" if c == "$" else "sq")
            word += "\0"
            i += 2 if c == "$" else 1
        elif c == '"':
            emit(c)
            modes.append("dq")
            word += "\0"
            i += 1
        elif c == "$" and nxt == "{":
            emit("${")
            modes.append("brace")
            word += "\0"
            i += 2
        elif c == "$" and nxt == "(":
            push_cmd("$(")
            word += "\0"
            i += 2
        elif c == "(":
            if case_state() == "pattern":
                emit(c)
            else:
                push_cmd(c)
            i += 1
        elif c == ")":
            if case_state() == "pattern":
                top.cases[-1] = "body"
            elif top.closer == ")":
                pop()
            emit(c)
            i += 1
        elif c == "{" and not word and nxt in (" ", "\t", "\n") and top.stages and not "".join(top.stage).strip():
            push_cmd(c, "}")
            i += 1
        elif c == "}" and not word and top.closer == "}" and (nxt == "" or nxt in BOUNDARY) and command_position(""):
            pop()
            emit(c)
            i += 1
        elif c == "|":
            if nxt == "|":
                finish(top)
                emit("||")
                i += 2
            elif case_state() == "pattern":
                emit(c)
                i += 1
            else:
                top.stages.append((top.start, "".join(top.stage).strip()))
                emit(c)
                top.stage, top.start = [], line
                i += 2 if nxt == "&" else 1
        elif c == ";":
            if src.startswith(";;", i) or src.startswith(";&", i):
                if case_state() == "body":
                    top.cases[-1] = "pattern"
                k = 3 if src.startswith(";;&", i) else 2
            else:
                k = 1
            finish(top)
            emit(src[i : i + k])
            i += k
        elif c == "&":
            if nxt == "&":
                finish(top)
                emit("&&")
                i += 2
            elif nxt == ">" or (i > 0 and src[i - 1] in "<>"):
                emit(c)
                i += 1
            else:
                finish(top)
                emit(c)
                i += 1
        elif c == "<" and src.startswith("<<", i):
            m = None if src.startswith("<<<", i) else HEREDOC.match(src, i)
            if m:
                heredocs.append((m.group(3), m.group(1) == "-"))
                emit(m.group(0))
                i = m.end()
            else:
                k = 3 if src.startswith("<<<", i) else 2
                emit(src[i : i + k])
                i += k
        elif c == "\n":
            line += 1
            i += 1
            for term, dash in heredocs:
                while i < n:
                    j = src.find("\n", i)
                    body = src[i : n if j < 0 else j]
                    i = n if j < 0 else j + 1
                    line += 1
                    if (body.lstrip("\t") if dash else body) == term:
                        break
            heredocs = []
            if not (top.stages and not "".join(top.stage).strip()):
                finish(top)
            emit("\n")
        else:
            emit(c)
            if c not in BOUNDARY:
                word += c
            i += 1

    if word:
        keyword(word)
    if len(frames) != 1 or modes != ["cmd"]:
        raise ValueError(f"unbalanced shell lexing at EOF: modes={modes}")
    finish(frames[0])
    return found, loops


def pipelines(src):
    """返回 bash 源码中每条多级管道各级的 [(起始行号, 文本), ...]。"""
    return [stages for _, stages in _lex(src)[0] if len(stages) > 1]


def _grep_exits_early(words):
    for w in words[1:]:
        w = w.strip("'\"")
        if w == "--":
            break
        if w.startswith("--"):
            if w.split("=")[0] in ("--quiet", "--silent", "--max-count"):
                return True
        elif w.startswith("-") and len(w) > 1:
            for ch in w[1:]:
                if ch in "qm":
                    return True
                if ch in GREP_ARG_OPTS:
                    break
    return False


def _mapfile_exits_early(words):
    """mapfile/readarray 带 -n（-n<N>、-n <N>、选项簇内的 n）或 --n/--numlines 时只读 N 行即停止。"""
    skip = False
    for w in words[1:]:
        w = w.strip("'\"")
        if skip:
            skip = False
            continue
        # bash 内建在首个操作数处停止选项解析：`mapfile arr -n 1` 的 -n 不生效、读满输入，不是提前退出。
        if w == "--" or not w.startswith("-") or len(w) == 1:
            break
        if w.startswith("--"):
            if w.split("=")[0] in ("--n", "--numlines"):
                return True
            continue
        for k, ch in enumerate(w[1:], 1):
            if ch == "n":
                return True
            if ch in MAPFILE_ARG_OPTS:
                skip = k == len(w) - 1
                break
    return False


def _awk_exits_early(prog):
    """awk 程序在 END 块之外含 exit 或 nextfile（含 BEGIN 内），即可能在读完输入前停止读取。

    单引号程序内的 awk 字符串字面量先剥去（`print "exit"` 不是语句）。
    """
    prog = SHELL_SQ.sub(lambda q: AWK_STRING.sub('""', q.group(0)), prog)
    while m := AWK_END.search(prog):
        depth, j = 0, m.end() - 1
        while j < len(prog):
            depth += {"{": 1, "}": -1}.get(prog[j], 0)
            j += 1
            if depth == 0:
                break
        prog = prog[: m.start()] + prog[j:]
    return bool(AWK_EXIT.search(prog))


def _group_exits_early(group):
    """( ... ) 或 { ...; } 内任一命令列表的首级（读组 stdin 者）是提前退出的读取器。"""
    end = group.rfind(")" if group[0] == "(" else "}")
    inner = group[1:end] if end > 0 else group[1:]
    return any(
        early_exit_reader(stages[0][1].lstrip(";&| \t\n")) for depth, stages in _lex(inner)[0] if depth == 0
    )


def _skip_wrapper_args(cmd, s):
    """跳过包装命令的选项（含带独立参数者）与被包装命令前的操作数，返回被包装命令起的文本。"""
    arg_opts, operands = WRAPPERS[cmd]
    opts = True
    while m := WORD.match(s):
        w = m.group(1)
        if opts and w == "--":
            opts = False
        elif opts and w.startswith("-") and len(w) > 1:
            if w in arg_opts:
                s = s[m.end() :]
                m = WORD.match(s)
                if not m:
                    return ""
        elif operands:
            opts, operands = False, operands - 1
        else:
            return s
        s = s[m.end() :]
    return s


def _until_condition_reads(cond):
    """until 条件含读取器：首次读取成功即结束循环。! 取反的条件只在读取失败（EOF）时成立，与 while 同样读满输入。"""
    for _, stages in cond:
        texts = [text.lstrip(";&| \t\n") for _, text in stages]
        if texts[0].split(None, 1)[:1] == ["!"]:
            continue
        if any(early_exit_reader(text) for text in texts):
            return True
    return False


def early_exit_reader(stage):
    """读取级读到结论后是否会提前关闭 stdin（READER_FAMILIES：grep -q/-m、head、sed q、awk exit/nextfile、read、
    mapfile/readarray -n、以 break/exit 结束或 until 条件含读取器的 while/until/select）。

    先剥去 VAR=value、!/time/command/builtin/env/exec 前缀与 nohup/timeout/nice/stdbuf 包装（连同其选项与 timeout
    时长）、取命令基名；子壳或花括组展开后逐条判定。
    """
    s = stage.strip()
    while s:
        if s[0] in "({":
            return _group_exits_early(s)
        m = WORD.match(s)
        cmd = m.group(1).rsplit("/", 1)[-1]
        if not (ASSIGNMENT.match(m.group(1)) or cmd in PREFIXES or cmd in WRAPPERS):
            break
        s = s[m.end() :]
        if cmd in WRAPPERS:
            s = _skip_wrapper_args(cmd, s)
    if not s:
        return False
    if cmd in LOOPS:
        return any(
            broken or (kind == "until" and _until_condition_reads(cond))
            for depth, kind, broken, cond in _lex(s)[1]
            if depth == 1
        )
    words = [cmd, *s[m.end() :].split()]
    rest = " ".join(words[1:])
    if cmd in ("grep", "egrep", "fgrep"):
        return _grep_exits_early(words)
    if cmd in ("head", "read"):
        return True
    if cmd in ("mapfile", "readarray"):
        return _mapfile_exits_early(words)
    if cmd == "sed":
        return bool(SED_QUIT.search(rest))
    if cmd in ("awk", "gawk", "mawk", "nawk"):
        return _awk_exits_early(rest)
    return False


def hazards(src):
    """各管道第二级起（读 pipe 的各级）中提前退出的读取器，返回 (行号, 该级文本)。"""
    return [
        (line, text)
        for stages in pipelines(src)
        for line, text in stages[1:]
        if early_exit_reader(text)
    ]


class ShellPipeHazardTests(unittest.TestCase):
    def test_detector_flags_synthetic_hazard(self):
        found = hazards("set -euo pipefail\nseq 1 200000 | grep -q 1\n")
        self.assertEqual([line for line, _ in found], [2], found)

    def test_detector_flags_hazard_shapes(self):
        for snippet in (
            'if ! seq 1 200000 | grep -q 1; then echo miss; fi',
            'x="$(sed -n \'s/^k=//p\' "$f" | head -1)"',
            'if ! "$bin" serve -h 2>&1 | grep -Eq -- "(^|[[:space:]])-{1,2}config"; then :; fi',
            'echo "${r:+($(printf \'%s\' "$r" | head -c 120))}"',
            'find . -type f \\\n  | grep -m1 x',
            "seq 1 200000 | sed -n '/1/{p;q;}'",
            'seq 1 200000 | grep -qiE "a|b" || exit 1',
        ):
            with self.subTest(snippet=snippet):
                self.assertTrue(hazards(snippet + "\n"), snippet)

    def test_detector_flags_newly_covered_shapes(self):
        for snippet, stage in (
            ("seq 1 200000 | head -1 | sed 's/^/x/'", "head -1"),
            ("x=\"$(seq 1 200000 | head -n 3 | tr -d ' ')\"", "head -n 3"),
            ("seq 1 200000 | awk '{print $1; exit}'", "awk '{print $1; exit}'"),
            ("seq 1 200000 | awk 'BEGIN { exit 1 }'", "awk 'BEGIN { exit 1 }'"),
            ("seq 1 200000 | awk '/7/{f=1; exit} END{print f}' | cat", "awk '/7/{f=1; exit} END{print f}'"),
            ("seq 1 200000 | read -r first", "read -r first"),
            ("seq 1 200000 | IFS= read -r first", "IFS= read -r first"),
            ("if seq 1 200000 | LC_ALL=C grep -q 1; then :; fi", "LC_ALL=C grep -q 1"),
            ("seq 1 200000 | env LC_ALL=C grep -m1 1 | wc -l", "env LC_ALL=C grep -m1 1"),
            ("seq 1 200000 | grep -q 1 | cat", "grep -q 1"),
        ):
            with self.subTest(snippet=snippet):
                self.assertEqual(hazards(snippet + "\n"), [(1, stage)], snippet)

    def test_wrapped_and_prefixed_readers_flagged_and_harmful(self):
        """子壳/花括组、awk nextfile、带 break 的 while read、绝对路径与 exec 前缀：检测器命中，且 pipefail 下确实报 141。"""
        grep = shutil.which("grep") or "/usr/bin/grep"
        shapes = (
            ("seq 1 200000 | (head -1)", "(head -1)"),
            ("seq 1 200000 | ( head -1 )", "( head -1 )"),
            ("seq 1 200000 | { head -1; }", "{ head -1; }"),
            ("seq 1 200000 | awk '$1 == 7 { print; nextfile }'", "awk '$1 == 7 { print; nextfile }'"),
            ("seq 1 200000 | while read -r l; do break; done", "while read -r l; do break; done"),
            (f"seq 1 200000 | {grep} -q 1", f"{grep} -q 1"),
            ("seq 1 200000 | exec grep -q 1", "exec grep -q 1"),
        )
        with tempfile.TemporaryDirectory(dir="/tmp") as tmp:
            for snippet, stage in shapes:
                with self.subTest(snippet=snippet):
                    self.assertEqual(hazards(snippet + "\n"), [(1, stage)], snippet)
                    script = Path(tmp) / "shape.sh"
                    script.write_text(f"set -euo pipefail\n{snippet}\necho survived\n", encoding="utf-8")
                    run = subprocess.run(["bash", str(script)], capture_output=True, text=True)
                    self.assertEqual(run.returncode, 141, run)
                    self.assertNotIn("survived", run.stdout)

    def test_detector_flags_group_loop_and_prefix_variants(self):
        for snippet, stage in (
            ("seq 1 200000 | {\n  head -1\n} | cat", "{\n  head -1\n}"),
            ("if seq 1 200000 | (grep -q 1); then :; fi", "(grep -q 1)"),
            ("seq 1 200000 | (cd /tmp && head -1)", "(cd /tmp && head -1)"),
            ("seq 1 200000 | while read -r l\ndo\n  [ \"$l\" = 7 ] && break\ndone", "while read -r l\ndo\n  [ \"$l\" = 7 ] && break\ndone"),
            ("seq 1 200000 | until ! read -r l; do break; done", "until ! read -r l; do break; done"),
            ("seq 1 200000 | while read -r l; do for x in a; do break 2; done; done", "while read -r l; do for x in a; do break 2; done; done"),
            ("seq 1 200000 | /usr/bin/env LC_ALL=C grep -q 1", "/usr/bin/env LC_ALL=C grep -q 1"),
            ("seq 1 200000 | exec /usr/bin/head -1", "exec /usr/bin/head -1"),
            ("seq 1 200000 | until IFS= read -r l\ndo\n  :\ndone", "until IFS= read -r l\ndo\n  :\ndone"),
            ("seq 1 200000 | until read -r a && read -r b; do :; done", "until read -r a && read -r b; do :; done"),
            ("seq 1 200000 | until [ \"$(head -1)\" = 7 ]; do :; done", "until [ \"$(head -1)\" = 7 ]; do :; done"),
            ("seq 1 200000 | while read -r l; do { exit 1; }; done", "while read -r l; do { exit 1; }; done"),
            ("seq 1 200000 | while read -r l; do if [ -z \"$l\" ]; then exit 1; fi; done",
             "while read -r l; do if [ -z \"$l\" ]; then exit 1; fi; done"),
            ("seq 1 200000 | while read -r l; do case $l in 7) break ;; esac; done",
             "while read -r l; do case $l in 7) break ;; esac; done"),
            ("seq 1 200000 | select x in a; do [ -n \"$x\" ] && exit; done", "select x in a; do [ -n \"$x\" ] && exit; done"),
            ("seq 1 200000 | timeout -s KILL 5 grep -q 1", "timeout -s KILL 5 grep -q 1"),
            ("seq 1 200000 | timeout --foreground 5 head -1", "timeout --foreground 5 head -1"),
            ("seq 1 200000 | env LC_ALL=C /usr/bin/timeout 5 head -1", "env LC_ALL=C /usr/bin/timeout 5 head -1"),
        ):
            with self.subTest(snippet=snippet):
                self.assertEqual(hazards(snippet + "\n"), [(1, stage)], snippet)

    def test_loop_family_and_wrappers_flagged_and_harmful(self):
        """until 条件读取、循环内 exit、select break、nohup/timeout/nice/stdbuf 包装：检测器命中且 pipefail 下报 141；
        对应的读满输入写法不命中且照常完成。"""
        shapes = (
            "until read -r l; do :; done",
            "until grep -q 1; do :; done",
            "while read -r l; do exit; done",
            "while read -r l; do [ \"$l\" = 7 ] && exit; done",
            "while read -r l; do case $l in 7) exit ;; esac; done",
            "until false; do read -r l; exit; done",
            "select x in a; do break; done",
            "timeout 5 head -1",
            "nohup head -1",
            "nice -n 5 head -1",
            "stdbuf -oL grep -q 1",
        )
        safe = (
            "while read -r l; do :; done",
            "while grep -q 1; do :; done",
            "until ! read -r l; do :; done",
            # select 读到 EOF 时状态为 1；断言生产者未收 SIGPIPE。
            "select x in a; do :; done >/dev/null || [ \"${PIPESTATUS[0]}\" = 0 ]",
            "awk '{print \"exit\"}' >/dev/null",
            "mapfile arr -n 1",
            "timeout 5 cat >/dev/null",
        )
        with tempfile.TemporaryDirectory(dir="/tmp") as tmp:
            for shape in shapes:
                snippet = f"seq 1 200000 | {shape}"
                with self.subTest(snippet=snippet):
                    self.assertEqual(hazards(snippet + "\n"), [(1, shape)], snippet)
                    script = Path(tmp) / "shape.sh"
                    script.write_text(f"set -euo pipefail\n{snippet}\necho survived\n", encoding="utf-8")
                    run = subprocess.run(["bash", str(script)], capture_output=True, text=True)
                    self.assertEqual(run.returncode, 141, run)
                    self.assertNotIn("survived", run.stdout)
            script = Path(tmp) / "safe.sh"
            script.write_text(
                "set -euo pipefail\n" + "".join(f"seq 1 20000 | {s}\n" for s in safe) + "echo survived\n",
                encoding="utf-8",
            )
            run = subprocess.run(["bash", str(script)], capture_output=True, text=True)
            self.assertEqual((run.returncode, run.stdout), (0, "survived\n"), run)
            self.assertEqual(hazards(script.read_text(encoding="utf-8")), [])

    def test_counted_readers_flagged_and_harmful(self):
        """mapfile/readarray -n 与 read -n：检测器命中，且 pipefail 下脚本中止（读满 N 后退出报 141，长选项被拒报非零）。"""
        shapes = (
            ("seq 1 200000 | mapfile -n 1 arr", "mapfile -n 1 arr", 141),
            ("seq 1 200000 | readarray --numlines 1 arr", "readarray --numlines 1 arr", None),
            ("seq 1 200000 | read -n 5 -r line", "read -n 5 -r line", 141),
        )
        with tempfile.TemporaryDirectory(dir="/tmp") as tmp:
            for snippet, stage, rc in shapes:
                with self.subTest(snippet=snippet):
                    self.assertEqual(hazards(snippet + "\n"), [(1, stage)], snippet)
                    script = Path(tmp) / "shape.sh"
                    script.write_text(f"set -euo pipefail\n{snippet}\necho survived\n", encoding="utf-8")
                    run = subprocess.run(["bash", str(script)], capture_output=True, text=True)
                    self.assertNotEqual(run.returncode, 0, run)
                    if rc is not None:
                        self.assertEqual(run.returncode, rc, run)
                    self.assertNotIn("survived", run.stdout)
            script = Path(tmp) / "safe.sh"
            script.write_text(
                "set -euo pipefail\nseq 1 200000 > in.txt\nmapfile arr < in.txt\nread -r line < in.txt\n"
                "seq 1 200000 | mapfile -t arr\necho \"survived ${#arr[@]} $line\"\n",
                encoding="utf-8",
            )
            run = subprocess.run(["bash", str(script)], capture_output=True, text=True, cwd=tmp)
            self.assertEqual((run.returncode, run.stdout), (0, "survived 200000 1\n"), run)
            self.assertEqual(hazards(script.read_text(encoding="utf-8")), [])

    def test_detector_flags_counted_reader_variants(self):
        for snippet, stage in (
            ("seq 1 200000 | mapfile -n1 arr", "mapfile -n1 arr"),
            ("seq 1 200000 | mapfile -t -n 3 arr", "mapfile -t -n 3 arr"),
            ("seq 1 200000 | readarray -tn2 arr", "readarray -tn2 arr"),
            ("seq 1 200000 | mapfile -u 0 -n 1 arr", "mapfile -u 0 -n 1 arr"),
            ("seq 1 200000 | readarray --n 1 arr", "readarray --n 1 arr"),
            ("seq 1 200000 | builtin mapfile -n 1 arr", "builtin mapfile -n 1 arr"),
            ("seq 1 200000 | { mapfile -n 1 arr; }", "{ mapfile -n 1 arr; }"),
            ("seq 1 200000 | read --chars 5 line", "read --chars 5 line"),
        ):
            with self.subTest(snippet=snippet):
                self.assertEqual(hazards(snippet + "\n"), [(1, stage)], snippet)

    def test_declared_reader_families_each_flagged(self):
        """READER_FAMILIES 是守护的契约：非空，且每个声明族至少有一个被检测器命中的合成用例。"""
        cases = {
            "grep": "seq 1 200000 | grep -q 1",
            "head": "seq 1 200000 | head -1",
            "sed": "seq 1 200000 | sed -n '/1/{p;q;}'",
            "awk": "seq 1 200000 | awk '{print; exit}'",
            "read": "seq 1 200000 | read -n 5 -r line",
            "mapfile": "seq 1 200000 | mapfile -n 1 arr",
            "while": (
                "seq 1 200000 | while read -r l; do break; done",
                "seq 1 200000 | while read -r l; do exit; done",
                "seq 1 200000 | until read -r l; do :; done",
                "seq 1 200000 | select x in a; do break; done",
            ),
        }
        self.assertTrue(READER_FAMILIES)
        for family in READER_FAMILIES:
            command = family[0]
            with self.subTest(family=command):
                self.assertEqual(len(family), 3)
                self.assertTrue(all(family), family)
                self.assertIn(command, cases, "declared family lacks a synthetic case")
                snippets = cases[command] if isinstance(cases[command], tuple) else (cases[command],)
                for snippet in snippets:
                    self.assertTrue(hazards(snippet + "\n"), snippet)

    def test_detector_accepts_safe_forms(self):
        src = "\n".join((
            "set -euo pipefail",
            "# seq 1 200000 | grep -q 1",
            "grep -q 1 < <(seq 1 200000)",
            'grep -q x <<<"$v"',
            "x=\"$(sed -n '/^k=/{s///p;q;}' \"$f\")\"",
            "x=\"$(sed -n '/^k=/{s///p;q;}' < <(seq 1 200000))\"",
            "awk '{print $1; exit}' < <(seq 1 200000)",
            "head -c 160 < <(seq 1 200000)",
            "ps -eo comm | sed -n '1,12p'",
            "seq 1 200000 | sed -n '1,12p' | tr -d ' '",
            "seq 1 200000 | awk '{n++} END {exit !n}'",
            "seq 1 200000 | awk '$1 == 7 { f = 1 } END { exit !f }' && echo hit",
            "seq 1 200000 | while read -r l; do :; done",
            "seq 1 200000 | while read -r l; do [ \"$l\" = 7 ] && continue; done",
            "seq 1 200000 | while read -r l; do for x in a b; do break; done; done",
            "while read -r l; do break; done < <(seq 1 200000)",
            "seq 1 200000 | while grep -q 1; do :; done",
            "seq 1 200000 | until ! read -r l; do :; done",
            "seq 1 200000 | while read -r l; do (exit 1); done",
            "seq 1 200000 | while read -r l; do x=$(exit 1); done",
            "seq 1 200000 | while read -r l; do echo exit; done",
            "seq 1 200000 | while read -r l; do case $l in 7) echo exit ;; esac; done",
            "seq 1 200000 | select x in a; do :; done",
            "seq 1 200000 | until [ -n \"$x\" ]; do x=1; done",
            "seq 1 200000 | awk '{print \"exit\"}'",
            "seq 1 200000 | mapfile arr -n 1",
            "seq 1 200000 | timeout 5 cat",
            "seq 1 200000 | nohup cat",
            "seq 1 200000 | stdbuf -oL sed -n p",
            "until read -r l; do :; done < file",
            "seq 1 200000 | awk '{n++} END { nextfile }'",
            "seq 1 200000 | (cat)",
            "seq 1 200000 | { cat; }",
            "seq 1 200000 | env FOO=1 command cat",
            "seq 1 200000 | time builtin cat >/dev/null",
            "seq 1 200000 | LC_ALL=C grep -c 1",
            "mapfile arr < file",
            "read -r line",
            "read -n 5 -r line < file",
            "seq 1 200000 | mapfile arr",
            "seq 1 200000 | mapfile -t arr",
            "seq 1 200000 | readarray -u 0 -t arr",
            "seq 1 200000 | mapfile -d '' arr",
            "seq 1 200000 | mapfile -t -- arr",
            "case \"$w\" in --auth|--config) grep -q x <<<\"$w\" ;; esac",
            "cat <<'EOF'",
            "seq 1 200000 | grep -q 1",
            "EOF",
            "",
        ))
        self.assertEqual(hazards(src), [])

    def test_discovery_covers_managed_scripts(self):
        found = managed_scripts()
        for rel in (
            "scripts/svc.sh",
            "scripts/task.sh",
            "scripts/e2e.sh",
            "scripts/e2e-duties.sh",
            "scripts/lint-boundaries.sh",
            "scripts/hooks/pre-commit",
            "scripts/task",
            "scripts/dm",
            "scripts/agent-exec",
            "docker/entrypoint.sh",
            *EXTRA_SCRIPTS,
        ):
            self.assertIn(rel, found)
        for rel in ("scripts/evidence.py", "scripts/lint-imports.go", "docker/Dockerfile", "docker/README.md"):
            self.assertNotIn(rel, found)
        self.assertFalse([p for p in found if p.startswith("runtime/")], found)
        self.assertFalse(set(EXEMPT) - set(found), "stale EXEMPT entries")

    def test_managed_scripts_have_no_early_exit_pipe_reader(self):
        for rel in managed_scripts():
            if rel in EXEMPT:
                continue
            with self.subTest(script=rel):
                path = ROOT / rel
                self.assertTrue(path.is_file(), f"{rel} missing")
                found = hazards(path.read_text(encoding="utf-8"))
                self.assertEqual(
                    found, [], "\n".join(f"{rel}:{line}: | {text}" for line, text in found)
                )


if __name__ == "__main__":
    unittest.main()