host: 验测命令白名单 + 结构化 verify_argv（shell=True 收口）

方案 A（白名单）+ 方案 B-lite（结构化 argv）双轨：
- 分类器 classify_verify_command：POSIX 引号状态机扫描，
  simple（无 shell 形态）/ simple-env（仅 VAR=val 前缀）走
  shell=False，shell（白名单内形态）走唯一的 shell=True 门，
  denied（嵌套 shell、命令替换、未列首词、绝对/越界重定向、
  多行等）拒绝执行并审计，返回 126 按验测失败处理。
- 验测线 shell=True 从 4 处收口到 1 处（run_gated_verify），
  run_verify 与轮次验收 2 处全部改走该门；bandit High 4->2
 （nosec 0，未做任何压制）。
- verify_argv：merge/goal/goal-resume 的 --verify-argv
  （REMAINDER，须放最后；空数组报 usage 退出，不静默跳过），
  任务记录 verify_argv，goal-resume 原样重放；goal-resume
  的 max_rounds==0 钳制计入 argv 形态。
- verify-audit：扫描 state 旧任务并分类；--apply 只转 simple，
  shell/denied 只列清单；审计本身不启动 codex server。
- notify_failure 不动：算子自有配置 + 插值已 shlex.quote
  （main 已有），白名单不适用，见 PR 正文定性表。
测试：新增 tests/test_host_shell_policy.py 30 项（含 naive
shlex.split 破坏回归锁、拒绝不执行标记文件、merge 回滚、
goal unmet/argv 达标、CLI REMAINDER/空数组、verify-audit
端到端）；顶层 135、deploy 39、cloud 1430（skipped 4）全过。

host.py

@@ -34,6 +34,7 @@ turn/completed 通知，所以多个 thread 的 turn 可以并发跑。
   python3 host.py delegate --parent-task add2 --task "只读检查 add2 有没有边界问题" --name add2-review
   python3 host.py goal --repo work/host-repo --task "创建 result.txt，内容只有一行 goal-ok" \
       --verify "test -f result.txt" --name goal1
+  python3 host.py verify-audit        # 扫描旧任务的验收命令：分类 + 迁移清单
   python3 host.py parallel --jobs examples/host-jobs.json
   python3 host.py automations --add --name daily-check --repo work/host-repo \
       --task "只读检查 sample.py 有没有明显问题" --interval 86400
@@ -1369,19 +1370,613 @@ def detect_verify_command(repo):
     return None
 
 
-def run_verify(command, cwd):
-    if not command:
+# ---------------- 验测命令白名单（shell=True 收口） ----------------
+# 原来 4 处 shell=True：run_verify（合并后验测）、notify_failure
+# （失败通知模板）、轮次验收 2 处（goal 循环，超时/不限时两分支）。
+# 本节只收口“验测命令”这一条线；notify 见后文定性（算子自有
+# 配置 + 插值已 shlex.quote，不进白名单，代码不动）。
+#
+# 信任边界——先说清白名单不是沙箱，否则就是虚假安全感：
+#   - 验测命令的来源只有三处：算子 CLI --verify、任务创建时算子
+#     给的 verify_command（goal-resume 原样重放）、系统默认
+#     detect_verify_command（固定串，无插值）。模型输出、任务
+#     正文、仓库文件从不进入验测命令；run_verify/轮次验收内部
+#     没有任何插值。白名单防的是“被植入的旧任务记录/手误写出的
+#     越界命令”，不是“算子本人想跑的命令”。
+#   - 白名单是“形态”白名单：首词必须是验测/构建类程序或仓内
+#     可执行脚本，且不含嵌套 shell、命令替换、绝对/越界重定向、
+#     多行等逃逸形态。首词一旦是 python3/make/npm 这类运行器，
+#     它的参数（-c、Makefile、npm script）本身就能跑任意代码——
+#     这是验测的本意（测试本来就要跑仓库代码），白名单拦不住、
+#     也不声称拦住。拦住的是 curl/rm/sudo/sh 这类与验测无关的
+#     程序、以及借 shell 语法偷渡第二条命令的写法。
+#   - 执行策略：拒绝 = 阻塞 + 审计，绝不静默回退到旧 shell 路径。
+#     拒绝时按“验测失败”处理：merge 回滚、goal 记 unmet 并把
+#     拒绝原因喂回 agent（不是代码问题，是命令形态问题，算子
+#     须迁移命令）。迁移工具见 verify-audit（文末 main 分支）。
+#   - 结构化路径（方案 B 的一部分）：verify_argv 是 argv 数组，
+#     走 shell=False，不经过 shell 解析，也不经过白名单——因为
+#     没有 shell 形态可校验；它的安全含义只是“无 shell 解析惊
+#     喜”，不是“argv[0] 被白名单过”。算子显式给 argv 即表示
+#     对该 argv 负责。
+VERIFY_COMMAND_ALLOWLIST = frozenset({
+    # 测试/构建运行器
+    "python3", "python", "pytest", "node", "npm", "yarn", "pnpm", "bun",
+    "deno", "make", "go", "cargo", "ruby", "bundle", "rspec", "php",
+    "phpunit", "java", "mvn", "gradle",
+    # 验测里常见的最小谓词/输出工具（只读或只写 stdout）
+    "test", "[", "true", "false", "echo", "grep", "cat", "ls", "diff",
+    "wc", "head", "tail", "sleep",
+    # sleep 是惰性的（只等待，不读写、不执行），且受 verify_timeout
+    # 上限约束；验测里“等一会儿再查文件”（sleep N && test -f）
+    # 是常见写法，旧测试 test_goal_verify_timeout_is_a_verify_failure
+    # 也用它模拟长验测，故列入白名单而非改测试迁就实现。
+})
+# 刻意不在白名单内：sh/bash/dash/zsh（嵌套 shell）、env/sudo/su、
+# curl/wget/ssh/scp（外联）、rm/mv/cp/chmod（改动文件系统）、
+# pip/pip3（装包取码；python3 -m pip 同理不在首词层拦截，见上文
+# “白名单不是沙箱”）、cd/export/time（shell 内建，改 cwd/环境，
+# 请用 VAR=val cmd 形态或 host 自身的 cwd）。
+_VERIFY_DENIED_RC = 126
+
+
+def _scan_shell_command(command):
+    """扫描一条 shell 字符串，返回 (segments, features, denied)。
+    segments：每段 {"words": [{"text", "expand", "glob"}],
+    "redirects": [{"target", "expand", "glob", "fd"}]}；features 是
+    {"compound", "pipe", "redirect", "env", "glob", "assign"} 的子集
+    （assign 由 classify 在分段校验时补）；denied 非 None 时整条
+    拒绝，原因是给算子看的迁移提示。
+    扫描按 POSIX 引号状态机：单引号内一切字面；双引号内 $ 仍
+    展开、反引号仍执行；反斜杠转义下一字符（双引号内只对
+    $ ` " \\ 换行起转义，其余保留反斜杠——本扫描只关心元字符
+    是否被转义，取保守：反斜杠后的字符一律视为字面）。
+    """
+    if "\x00" in command:
+        return [], set(), "命令含 NUL 字节"
+    if "\n" in command or "\r" in command:
+        return [], set(), "命令含换行：多行命令请改写成 && 连接的单行"
+    segments = []
+    features = set()
+    words = []
+    redirects = []
+    cur = []
+    cur_expand = False
+    cur_glob = False
+    cur_started = False
+    pending_redirect = None  # None | "file" | "fd"（>& / <& 后）
+    i = 0
+    n = len(command)
+    state = "normal"  # normal | squote | dquote
+
+    def flush_word():
+        nonlocal cur, cur_expand, cur_glob, cur_started, pending_redirect
+        if not cur_started:
+            return
+        word = {"text": "".join(cur), "expand": cur_expand,
+                "glob": cur_glob}
+        if pending_redirect is not None:
+            redirects.append({"target": word["text"],
+                              "expand": word["expand"],
+                              "glob": word["glob"],
+                              "fd": pending_redirect == "fd"})
+            pending_redirect = None
+        else:
+            words.append(word)
+        cur = []
+        cur_expand = False
+        cur_glob = False
+        cur_started = False
+
+    def end_segment():
+        nonlocal words, redirects
+        flush_word()
+        if pending_redirect is not None:
+            return "重定向后缺目标"
+        segments.append({"words": words, "redirects": redirects})
+        words = []
+        redirects = []
+        return None
+
+    def dollar(i):
+        """处理 $ 后的展开：返回 (新下标, 拒绝原因或 None)。
+        调用方保证 command[i] == "$"。"""
+        nonlocal cur_started, cur_expand
+        nxt = command[i + 1] if i + 1 < n else ""
+        if nxt == "(":
+            return i, "命令替换 $(...) 不在白名单形态内"
+        if nxt == "{":
+            j = command.find("}", i + 2)
+            if j < 0:
+                return i, "${ 未闭合"
+            cur.append(command[i:j + 1])
+            cur_started = True
+            cur_expand = True
+            features.add("env")
+            return j + 1, None
+        if nxt and (nxt.isalnum() or nxt in "@*#?$!-"):
+            cur.append("$")
+            cur_started = True
+            cur_expand = True
+            features.add("env")
+            return i + 1, None
+        cur.append("$")
+        cur_started = True
+        return i + 1, None
+
+    while i < n:
+        c = command[i]
+        if state == "squote":
+            if c == "'":
+                state = "normal"
+            else:
+                cur.append(c)
+            i += 1
+            continue
+        if state == "dquote":
+            if c == '"':
+                state = "normal"
+                i += 1
+                continue
+            if c == "\\" and i + 1 < n and command[i + 1] in "$`\"\\\n":
+                cur.append(command[i + 1])
+                i += 2
+                continue
+            if c == "`":
+                return [], set(), "反引号命令替换不在白名单形态内"
+            if c == "$":
+                i, err = dollar(i)
+                if err:
+                    return [], set(), err
+                continue
+            cur.append(c)
+            i += 1
+            continue
+        # state == "normal"
+        if c in " \t":
+            flush_word()
+            i += 1
+            continue
+        if c == "\\" and i + 1 < n:
+            cur.append(command[i + 1])
+            cur_started = True
+            i += 2
+            continue
+        if c == "'":
+            state = "squote"
+            cur_started = True
+            i += 1
+            continue
+        if c == '"':
+            state = "dquote"
+            cur_started = True
+            i += 1
+            continue
+        if c == "`":
+            return [], set(), "反引号命令替换不在白名单形态内"
+        if c == "$":
+            i, err = dollar(i)
+            if err:
+                return [], set(), err
+            continue
+        if c == "(" or c == ")":
+            return [], set(), "子 shell/分组括号不在白名单形态内"
+        if c == "{" or c == "}":
+            return [], set(), "花括号分组/展开不在白名单形态内"
+        if c == "#":
+            if not cur_started:
+                return [], set(), "行内注释 # 会截断可见命令，请去掉注释再写"
+            cur.append(c)
+            cur_started = True
+            i += 1
+            continue
+        if c == "~":
+            if not cur_started or (cur and cur[-1] == "="):
+                cur_expand = True
+                features.add("env")
+            cur.append(c)
+            cur_started = True
+            i += 1
+            continue
+        if c in "*?":
+            cur_glob = True
+            features.add("glob")
+            cur.append(c)
+            cur_started = True
+            i += 1
+            continue
+        if c == "[":
+            # 段首（或仅跟赋值前缀后）且后跟空白的 [ 是 test 命令
+            # 本身，不是 glob；其它位置的 [ 是 glob 形态。
+            prefix = [w["text"] for w in words]
+            at_command = (not cur_started
+                          and (i + 1 >= n or command[i + 1] in " \t")
+                          and all(re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", t)
+                                  for t in prefix))
+            if at_command:
+                cur.append(c)
+                cur_started = True
+                i += 1
+                continue
+            cur_glob = True
+            features.add("glob")
+            cur.append(c)
+            cur_started = True
+            i += 1
+            continue
+        if c == "!":
+            if not words and not cur_started:
+                return [], set(), "取反 ! 不在白名单形态内"
+            cur.append(c)
+            cur_started = True
+            i += 1
+            continue
+        if c == "&":
+            if command[i + 1:i + 2] == ">":
+                flush_word()
+                features.add("redirect")
+                pending_redirect = "file"
+                i += 2
+                if command[i:i + 1] == ">":
+                    i += 1
+                continue
+            if command[i + 1:i + 2] == "&":
+                err = end_segment()
+                if err:
+                    return [], set(), err
+                features.add("compound")
+                i += 2
+                continue
+            return [], set(), "后台 & 不在白名单形态内（验收须前台跑完才有 rc）"
+        if c == "|":
+            if command[i + 1:i + 2] == "|":
+                err = end_segment()
+                if err:
+                    return [], set(), err
+                features.add("compound")
+                i += 2
+                continue
+            err = end_segment()
+            if err:
+                return [], set(), err
+            features.add("pipe")
+            i += 1
+            continue
+        if c == ";":
+            err = end_segment()
+            if err:
+                return [], set(), err
+            features.add("compound")
+            i += 1
+            continue
+        if c in "><":
+            # 数字前缀是 fd（如 2>、2>&1），不是参数。
+            if cur_started and "".join(cur).isdigit():
+                cur = []
+                cur_expand = False
+                cur_glob = False
+                cur_started = False
+            else:
+                flush_word()
+            if c == "<" and command[i + 1:i + 2] == "<":
+                return [], set(), "here-doc << 不在白名单形态内"
+            if command[i + 1:i + 2] == "&":
+                pending_redirect = "fd"
+                i += 2
+            else:
+                pending_redirect = "file"
+                if c == ">":
+                    i += 1
+                    if command[i:i + 1] == ">":
+                        i += 1
+                    if command[i:i + 1] == "|":
+                        i += 1
+                else:
+                    i += 1
+            features.add("redirect")
+            continue
+        cur.append(c)
+        cur_started = True
+        i += 1
+    if state != "normal":
+        return [], set(), "引号未闭合"
+    err = end_segment()
+    if err:
+        return [], set(), err
+    return segments, features, None
+
+
+def _is_assign_word(text):
+    return re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", text) is not None
+
+
+def classify_verify_command(command, cwd=None):
+    """验测字符串分类，返回 dict：
+    kind: "empty"（空/空白串，等价无验收）| "simple"（无 shell
+      形态，走 shell=False）| "simple-env"（只有 VAR=val 前缀，
+      走 shell=False + env）| "shell"（白名单内 shell 形态，走
+      shell=True）| "denied"（拒绝执行，reason 给迁移提示）；
+    argv: simple/simple-env 时的 argv 数组（shell/denied 时 None）；
+    env: simple-env 时的赋值 dict（其它 kind 时 None）；
+    reason: denied 时的拒绝原因，其它 kind 时 ""；
+    features: 形态特征集合；notes: 审计用备注（如脚本路径因无
+      cwd 未校验——执行时 cwd 一定有，以执行时结论为准）。
+    """
+    if command is None or not command.strip():
+        return {"kind": "empty", "argv": None, "env": None, "reason": "",
+                "features": frozenset(), "notes": []}
+    segments, features, denied = _scan_shell_command(command)
+    if denied is not None:
+        return {"kind": "denied", "argv": None, "env": None,
+                "reason": denied, "features": features, "notes": []}
+    notes = []
+    argv = None
+    env = None
+    for seg in segments:
+        words = seg["words"]
+        redirects = seg["redirects"]
+        if not words and not redirects:
+            return {"kind": "denied", "argv": None, "env": None,
+                    "reason": "空段：结尾操作符后缺命令",
+                    "features": features, "notes": notes}
+        if not words:
+            return {"kind": "denied", "argv": None, "env": None,
+                    "reason": "只有重定向、没有命令",
+                    "features": features, "notes": notes}
+        assigns = []
+        idx = 0
+        while idx < len(words) and _is_assign_word(words[idx]["text"]):
+            assigns.append(words[idx]["text"])
+            idx += 1
+        if assigns:
+            features = features | {"assign"}
+        if idx >= len(words):
+            return {"kind": "denied", "argv": None, "env": None,
+                    "reason": "只有环境变量赋值、没有命令",
+                    "features": features, "notes": notes}
+        cmd_word = words[idx]["text"]
+        if "/" in cmd_word:
+            if cmd_word.startswith("/"):
+                return {"kind": "denied", "argv": None, "env": None,
+                        "reason": f"绝对路径命令不在白名单内：{cmd_word}",
+                        "features": features, "notes": notes}
+            if ".." in cmd_word.split("/"):
+                return {"kind": "denied", "argv": None, "env": None,
+                        "reason": f"越界路径命令不在白名单内：{cmd_word}",
+                        "features": features, "notes": notes}
+            if cwd is not None:
+                if not os.path.isdir(cwd):
+                    return {"kind": "denied", "argv": None, "env": None,
+                            "reason": "脚本路径无法校验：cwd 不存在",
+                            "features": features, "notes": notes}
+                base = os.path.realpath(cwd)
+                target = os.path.realpath(os.path.join(cwd, cmd_word))
+                if os.path.commonpath([base, target]) != base:
+                    return {"kind": "denied", "argv": None, "env": None,
+                            "reason": f"脚本路径越出 cwd：{cmd_word}",
+                            "features": features, "notes": notes}
+                if not os.path.isfile(target) or not os.access(target, os.X_OK):
+                    return {"kind": "denied", "argv": None, "env": None,
+                            "reason": f"仓内脚本不存在或不可执行：{cmd_word}",
+                            "features": features, "notes": notes}
+            else:
+                notes.append(f"脚本路径未校验（无 cwd）：{cmd_word}")
+        elif cmd_word not in VERIFY_COMMAND_ALLOWLIST:
+            return {"kind": "denied", "argv": None, "env": None,
+                    "reason": f"首词不在验测白名单内：{cmd_word}",
+                    "features": features, "notes": notes}
+        for red in redirects:
+            target = red["target"]
+            is_fd = red["fd"] and (target.isdigit() or target == "-")
+            if is_fd:
+                continue
+            # >&file 这种写法在 shell 里是写文件，不是 fd 跳转：
+            # fd 标记但目标不是数字/- 时按文件目标校验。
+            if red["expand"] or red["glob"]:
+                return {"kind": "denied", "argv": None, "env": None,
+                        "reason": "重定向目标含变量展开/glob，无法静态校验",
+                        "features": features, "notes": notes}
+            if target == "/dev/null":
+                continue
+            if target.startswith("/") or ".." in target.split("/"):
+                return {"kind": "denied", "argv": None, "env": None,
+                        "reason": f"重定向目标不在 cwd 内：{target}",
+                        "features": features, "notes": notes}
+    if not features:
+        argv = [w["text"] for w in segments[0]["words"]]
+        return {"kind": "simple", "argv": argv, "env": None, "reason": "",
+                "features": features, "notes": notes}
+    if features == {"assign"}:
+        seg = segments[0]
+        assigns = []
+        idx = 0
+        while idx < len(seg["words"]) and _is_assign_word(
+                seg["words"][idx]["text"]):
+            assigns.append(seg["words"][idx]["text"])
+            idx += 1
+        env = {}
+        for a in assigns:
+            name, value = a.split("=", 1)
+            env[name] = value
+        argv = [w["text"] for w in seg["words"][idx:]]
+        return {"kind": "simple-env", "argv": argv, "env": env, "reason": "",
+                "features": features, "notes": notes}
+    return {"kind": "shell", "argv": None, "env": None, "reason": "",
+            "features": features, "notes": notes}
+
+
+def _audit_verify(state_path, command, kind, rc):
+    """验测执行审计：只写 host-run.log（state 同目录），写失败
+    静默——审计不能反过来搞垮验测本身。"""
+    if not state_path:
+        return
+    try:
+        log_event(state_path,
+                  f"verify {kind} rc={rc} cmd={command[:200]}")
+    except Exception:
+        pass
+
+
+def run_verify_argv(argv, cwd, timeout=600, audit_state_path=None):
+    """结构化验测：argv 数组走 shell=False，无 shell 解析。
+    空数组等价无验收（0, ""）；非字符串元素是调用方 bug，抛错
+    不静默——调用方（merge/goal）只传 CLI 解析出的字符串数组。"""
+    if not argv:
         return 0, ""
-    v = subprocess.run(command, shell=True, cwd=cwd,
-                       capture_output=True, text=True, timeout=600)
+    if not isinstance(argv, (list, tuple)):
+        raise TypeError("verify_argv 须是字符串数组")
+    if not all(isinstance(a, str) for a in argv):
+        raise TypeError("verify_argv 须是字符串数组")
+    v = subprocess.run(list(argv), shell=False, cwd=cwd,
+                       capture_output=True, text=True, timeout=timeout)
+    _audit_verify(audit_state_path, shlex.join(list(argv)), "argv",
+                  v.returncode)
     return v.returncode, (v.stdout + v.stderr).strip()[:2000]
 
 
-def merge_task(repo, worktree, branch, task_name, verify_command=None):
+def run_gated_verify(command, cwd, timeout=None, audit_state_path=None):
+    """验测字符串的唯一执行口：run_verify 与轮次验收都走这里。
+    simple/simple-env → shell=False；白名单内 shell 形态 →
+    shell=True（这是验测线唯一的 shell=True）；拒绝 → 不执行、
+    记审计、返回 126，调用方按验测失败处理（merge 回滚、goal
+    记 unmet）。TimeoutExpired/OSError 原样抛出，调用方原有
+    的超时/中断处理不变。"""
+    if not command or not command.strip():
+        return 0, ""
+    info = classify_verify_command(command, cwd=cwd)
+    kind = info["kind"]
+    if kind == "denied":
+        _audit_verify(audit_state_path, command, "denied",
+                      _VERIFY_DENIED_RC)
+        return _VERIFY_DENIED_RC, (
+            f"验收命令被白名单拒绝，未执行：{info['reason']}；"
+            f"请改用白名单内命令、verify_argv，或跑 "
+            f"host.py verify-audit 看迁移清单")
+    if kind in ("simple", "simple-env"):
+        env = None
+        if info["env"] is not None:
+            env = dict(os.environ)
+            env.update(info["env"])
+        v = subprocess.run(info["argv"], shell=False, cwd=cwd,
+                           capture_output=True, text=True, timeout=timeout,
+                           env=env)
+        _audit_verify(audit_state_path, command, kind, v.returncode)
+        return v.returncode, (v.stdout + v.stderr).strip()[:2000]
+    # shell：白名单内形态，仍走 shell=True——这是本文件验测线
+    # 唯一剩下的 shell=True（notify 的那处是另一条线，见
+    # notify_failure 的定性，代码不动）。
+    v = subprocess.run(command, shell=True, cwd=cwd,
+                       capture_output=True, text=True, timeout=timeout)
+    _audit_verify(audit_state_path, command, "shell", v.returncode)
+    return v.returncode, (v.stdout + v.stderr).strip()[:2000]
+
+
+def run_verify(command, cwd, audit_state_path=None):
+    if not command:
+        return 0, ""
+    return run_gated_verify(command, cwd, timeout=600,
+                            audit_state_path=audit_state_path)
+
+
+def build_verify_audit_report(state):
+    """纯函数：扫描 state 里带验收命令的任务，返回报告行列表。
+    每行：task（任务名）、kind（classify 结果；已有 argv 的记
+    "argv"）、command（字符串原值截断 200）、features（排序后
+    的形态特征）、reason（denied 时）、argv（simple 时可迁移的
+    数组；simple-env 已走 shell=False 无需迁移，记 None）、
+    notes、applied（调用方 --apply 后填 True）。
+    没有验收命令的任务（新形态 goal）不进报告——报告只回答
+    “哪些旧任务带有 shell 验收命令”。"""
+    report = []
+    for name, t in (state.get("tasks") or {}).items():
+        if not isinstance(t, dict):
+            continue
+        argv = t.get("verify_argv")
+        if argv is not None:
+            report.append({"task": name, "kind": "argv",
+                           "command": None,
+                           "features": [], "reason": "",
+                           "argv": (list(argv) if isinstance(argv, list)
+                                    else None),
+                           "notes": [], "applied": False})
+            continue
+        cmd = t.get("verify_command")
+        if not cmd:
+            continue
+        cwd = t.get("cwd")
+        info = classify_verify_command(
+            cmd, cwd=cwd if (cwd and os.path.isdir(cwd)) else None)
+        kind = info["kind"]
+        migratable = info["argv"] if kind == "simple" else None
+        report.append({"task": name, "kind": kind,
+                       "command": cmd[:200],
+                       "features": sorted(info["features"]),
+                       "reason": info["reason"],
+                       "argv": migratable,
+                       "notes": list(info["notes"]),
+                       "applied": False})
+    return report
+
+
+def _cmd_verify_audit(args, state):
+    """host.py verify-audit：只读扫描 + 可选 --apply。
+    --apply 只转换 kind == "simple" 的任务（argv 与原字符串在
+    无 shell 形态时严格等价）；simple-env 已走 shell=False 无需
+    转；shell/denied 一律只列清单、不自动改——转换 shell 形态
+    必须算子逐条确认，绝不静默丢弃语义。"""
+    report = build_verify_audit_report(state)
+    applied = 0
+    if args.apply:
+        tasks = state.get("tasks") or {}
+        for row in report:
+            if row["kind"] == "simple" and row["argv"] is not None:
+                t = tasks.get(row["task"])
+                if isinstance(t, dict) and t.get("verify_argv") is None:
+                    t["verify_argv"] = row["argv"]
+                    row["applied"] = True
+                    applied += 1
+        if applied:
+            save_state(args.state, state)
+    if args.json:
+        print(json.dumps({"state": args.state, "applied": applied,
+                          "tasks": report},
+                         indent=2, ensure_ascii=False))
+    else:
+        print(f"verify-audit: state={args.state} 任务 {len(report)} 条")
+        for row in report:
+            feats = ",".join(row["features"]) or "-"
+            extra = ""
+            if row["reason"]:
+                extra = f" reason={row['reason']}"
+            if row["notes"]:
+                extra += " notes=" + ";".join(row["notes"])
+            if row["applied"]:
+                extra += " [已转 verify_argv]"
+            elif row["kind"] == "simple":
+                extra += f" [可转 argv={row['argv']}]"
+            print(f"  {row['task']}: {row['kind']} features={feats}{extra}")
+        counts = {}
+        for row in report:
+            counts[row["kind"]] = counts.get(row["kind"], 0) + 1
+        print(f"汇总：{counts}；--apply 已转 {applied} 条")
+        denied = counts.get("denied", 0)
+        if denied:
+            print(f"注意：denied {denied} 条下次执行时将被阻塞（不执行），"
+                  f"须先改写命令或改用 verify_argv")
+        if counts.get("shell", 0):
+            print("shell 形态的任务仍走 shell=True（白名单内形态），"
+                  "迁移到 verify_argv 须算子逐条确认语义等价")
+
+
+def merge_task(repo, worktree, branch, task_name, verify_command=None,
+               verify_argv=None, audit_state_path=None):
     """合并 + 验测 + 回滚（纯 git 操作，不依赖 server，可单测）。
     返回 (ok, message)：验测失败时 reset 回合并前 HEAD，分支和
-    worktree 保留给修复；验测通过时才删 worktree/分支。"""
-    if verify_command is None:
+    worktree 保留给修复；验测通过时才删 worktree/分支。
+    verify_argv（白名单轨）：非 None 时优先于 verify_command 字符串，
+    走 shell=False；空数组表示跳过验测（等价 --no-verify）。
+    verify_command 为 None 且无 argv 时才用系统默认。"""
+    if verify_argv is None and verify_command is None:
         verify_command = detect_verify_command(repo)
     pre_head = git(["rev-parse", "HEAD"], repo).stdout.strip()
     # 合并前先清掉 worktree 里的 __pycache__ / *.pyc：agent 跑
@@ -1449,12 +2044,23 @@ def merge_task(repo, worktree, branch, task_name, verify_command=None):
         # 自动解成功，继续走下面的验测（不 return，落到验测分支）
         print(f"注意：合并冲突用 -X ours 自动解（重复加同一段），"
               f"新增行 {len(added)} 行已核对都在主仓里。")
-    if verify_command:
-        rc, out = run_verify(verify_command, repo)
-        if rc != 0:
-            git(["reset", "--hard", pre_head], repo)
-            return False, (f"合并后验测失败，已回滚到 {pre_head[:8]}，"
-                           f"分支 {branch} 和 worktree 保留：\n{out[:1000]}")
+    verify_rc = None
+    verify_out = ""
+    verify_ran = False
+    if verify_argv is not None:
+        if verify_argv:
+            verify_rc, verify_out = run_verify_argv(
+                verify_argv, repo, timeout=600,
+                audit_state_path=audit_state_path)
+            verify_ran = True
+    elif verify_command:
+        verify_rc, verify_out = run_verify(
+            verify_command, repo, audit_state_path=audit_state_path)
+        verify_ran = True
+    if verify_ran and verify_rc != 0:
+        git(["reset", "--hard", pre_head], repo)
+        return False, (f"合并后验测失败，已回滚到 {pre_head[:8]}，"
+                       f"分支 {branch} 和 worktree 保留：\n{verify_out[:1000]}")
     git(["worktree", "remove", "--force", worktree], repo)
     git(["branch", "-d", branch], repo)
     return True, f"merged {branch} into {repo}"
@@ -2008,7 +2614,10 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
                   max_rounds, model, effort, sandbox, cfg=None,
                   timeout=None, verify_timeout=600, start_round=1,
                   last_verify="", last_turn_status=None, last_turn_text="",
-                  resume_note=None):
+                  resume_note=None, verify_argv=None):
+# 演练注 FLAG：签名同位扩展——main 侧 N-12 加 resume_note，
+# 源线加 verify_argv；正文两处各有独立使用（首轮前言 vs
+# 验测路径选择），并集保留两边，交互语义待用户定。
     """目标循环（长任务版，对齐 Codex App 的可续跑语义）。
 
     两种验收形态（2026-10-27 起，console goal 模式走新形态）：
@@ -2019,6 +2628,10 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
     一直跑到模型宣告、预算耗尽或用户取消为止（与 turn_timeout=0
     不限时同约定）。轮次提示里会把标记约定写明，模型才知道怎么
     宣告；同一 thread 保留历史，续轮提示可以很短。
+    verify_argv（白名单轨）：argv 数组时优先于 verify 字符串，
+    走 shell=False；两者都没有才走新形态。verify_argv 由 CLI
+    --verify-argv 或任务记录的 verify_argv 来，goal-resume 原样
+    重放；新开目标循环时 verify_argv=None 会清掉旧记录。
     每轮结束立即落盘（检查点），进程被杀 / VM 回收后可用
     goal-resume 从 goal_rounds_done 的下一轮续跑，而不是从头重来。
     与云端 worker 的解析契约（cloud/sandbox.py）：
@@ -2045,8 +2658,18 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
     时行为与旧版逐字相同；开新目标分支绝不传它（新目标自带
     goal_text，不需要也不接受断点备注）。
     """
+    # 验测形态（白名单轨）：verify_argv 优先于 verify 字符串；两者
+    # 都没有时走模型自决新形态。verify_command 存字符串原值（可
+    # 能为 None），verify_argv 存数组（None 表示未用，顺带清掉
+    # 旧任务残留的 argv，避免 goal-resume 重放过期 argv）。
+    # verify_text 是喂给 agent 看的验收命令：argv 时用 shlex.join
+    # 还原成可读的一行（执行仍走 argv，不经过 shell 解析）。
+    verify_text = (shlex.join(verify_argv)
+                   if verify_argv is not None else verify)
+    has_verify = verify_text is not None
     task["goal"] = goal_text
     task["verify_command"] = verify
+    task["verify_argv"] = verify_argv
     task["goal_max_rounds"] = int(max_rounds)
     task["goal_rounds_done"] = int(start_round) - 1
     task["goal_last_verify"] = last_verify
@@ -2103,7 +2726,7 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
             break
         if round_no == 1 and not last_verify and not last_turn_status:
             prompt = goal_text
-            if verify is None:
+            if not has_verify:
                 # 新形态首轮必须把标记约定写进提示，否则模型不知
                 # 道怎么宣告完成，循环永远等不到结束信号。
                 prompt = (f"{goal_text}\n\n"
@@ -2116,14 +2739,14 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
             prompt = (f"上一轮回合未完成（状态：{last_turn_status}），"
                       f"请继续完成目标，不要重复已完成的部分。\n"
                       f"目标：{goal_text}\n")
-            if verify is not None:
-                prompt += (f"验收命令：{verify}\n"
+            if has_verify:
+                prompt += (f"验收命令：{verify_text}\n"
                            f"上一轮验收输出：\n{last_verify}\n")
             else:
                 prompt += (f"确认目标已完全达成时，在回复末尾单独一行输出："
                            f"{GOAL_DONE_MARKER}。\n")
             prompt += f"上一轮部分文本：\n{last_turn_text[:1000]}"
-        elif verify is None:
+        elif not has_verify:
             # 新形态续轮：同一 thread 已有历史，只需说明“还没
             # 宣告达成、继续做”，并重申标记约定。
             prompt = (f"上一轮未宣告目标达成，请继续完成目标，"
@@ -2137,7 +2760,7 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
             # 必须把验收命令和失败输出喂回去，才是可用的目标循环。
             prompt = (f"上一轮验收未通过，请针对性修复，不要重复已完成的部分。\n"
                       f"目标：{goal_text}\n"
-                      f"验收命令：{verify}\n"
+                      f"验收命令：{verify_text}\n"
                       f"上一轮验收输出：\n{last_verify}")
         # N-12：断点备注只贴第一轮 prompt（见 _first_turn 注释），
         # 绝不写回 goal_text——task["goal"] 在循环开始时已按原
@@ -2163,23 +2786,26 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
                     Exception("连续 3 轮回合失败，停止空转"))
         else:
             consecutive_failed = 0
-        if verify is None:
+        if not has_verify:
             # 新形态没有验收命令可跑，v_rc 记 -（轮次行格式不变，
             # 云端解析只读 verify_out 做 unmet 详情，不受影响）。
             v_rc, v_out = None, ""
         else:
             try:
                 v_rc, v_out = None, None
-                if verify_timeout and verify_timeout > 0:
-                    v = subprocess.run(verify, shell=True, cwd=task["cwd"],
-                                       capture_output=True, text=True,
-                                       timeout=verify_timeout)
+                vt = verify_timeout if (verify_timeout
+                                        and verify_timeout > 0) else None
+                if verify_argv is not None:
+                    v_rc, v_out = run_verify_argv(
+                        verify_argv, task["cwd"], timeout=vt,
+                        audit_state_path=state_path)
                 else:
-                    v = subprocess.run(verify, shell=True, cwd=task["cwd"],
-                                       capture_output=True, text=True)
-                v_rc = v.returncode
-                v_out = (v.stdout + v.stderr).strip()[:1000] or \
-                    f"exit code {v.returncode}"
+                    v_rc, v_out = run_gated_verify(
+                        verify, task["cwd"], timeout=vt,
+                        audit_state_path=state_path)
+                # 轮次行/goal_last_verify 保持旧口径：截断 1000，
+                # 空输出时记 exit code（旧代码同款，不改契约）。
+                v_out = v_out[:1000] or f"exit code {v_rc}"
             except subprocess.TimeoutExpired as exc:
                 # 验测本身也可能很长（测试套件）：超时是验测失败，
                 # 不是进程挂死，喂回下一轮让 agent 缩小验测范围。
@@ -2227,7 +2853,7 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
                     ensure_ascii=False) + "\n")
         except Exception:
             pass
-        if verify is None:
+        if not has_verify:
             # 模型自决：只有完整跑完的回合里的标记行才算数。
             if (result["status"] == "completed"
                     and _llm_declared_done(last_turn_text)):
@@ -2371,6 +2997,9 @@ def main():
     p.add_argument("--verify", default=None,
                    help="合并后验测命令；不传时仓库有 tests/ 就默认跑 unittest")
     p.add_argument("--no-verify", action="store_true")
+    p.add_argument("--verify-argv", nargs=argparse.REMAINDER, default=None,
+                   help="合并后验测 argv（优先于 --verify，走 shell=False；"
+                        "其后全部参数都作为 argv，须放在命令最后）")
     p = sub.add_parser("review-comment")
     p.add_argument("--task", required=True)
     p.add_argument("--file", required=True)
@@ -2453,6 +3082,9 @@ def main():
     p.add_argument("--verify", default=None,
                    help="验收命令；不传时走模型自决：模型在回合末尾"
                         "单独一行输出“目标已达成”即达标")
+    p.add_argument("--verify-argv", nargs=argparse.REMAINDER, default=None,
+                   help="验收 argv（优先于 --verify，走 shell=False；"
+                        "其后全部参数都作为 argv，须放在命令最后）")
     p.add_argument("--max-rounds", type=int, default=3,
                    help="轮次上限；0 表示不限轮次（模型宣告完成才结束），"
                         "默认 3")
@@ -2474,8 +3106,17 @@ def main():
                         "轮次与验收；与 --goal-text 同传时忽略（开新目标分支不用它）")
     p.add_argument("--verify-timeout", type=int, default=600,
                    help="验测命令上限（秒）；0 表示不限时，默认 600")
+    p.add_argument("--verify-argv", nargs=argparse.REMAINDER, default=None,
+                   help="断点续跑时覆盖任务的验收 argv（走 shell=False；"
+                        "其后全部参数都作为 argv，须放在命令最后）")
     p = sub.add_parser("parallel")
     p.add_argument("--jobs", required=True)
+    p = sub.add_parser("verify-audit")
+    p.add_argument("--apply", action="store_true",
+                   help="只转换 simple 形态的任务为 verify_argv；"
+                        "shell/denied 只列清单、不自动改")
+    p.add_argument("--json", action="store_true",
+                   help="以 JSON 输出报告（默认文本报告）")
     p = sub.add_parser("automations")
     p.add_argument("--add", action="store_true")
     p.add_argument("--name")
@@ -2879,6 +3520,18 @@ def main():
             pass
         print("host daemon stopped", flush=True)
         return
+    # verify-audit 是只读扫描工具（+ 可选 --apply 写回），绝不
+    # 启动 codex server：放在 srv 构造之前早返回，否则每次审计
+    # 都要拉起 app-server 进程。
+    # --verify-argv 用 REMAINDER：空数组表示算子写了标志却没给
+    # 参数，那是笔误不是“跳过验测”，直接报 usage 退出（exit 2），
+    # 绝不能静默当成 --no-verify。
+    if (args.cmd in ("merge", "goal", "goal-resume")
+            and getattr(args, "verify_argv", None) == []):
+        ap.error("--verify-argv 后须至少给一个参数")
+    if args.cmd == "verify-audit":
+        _cmd_verify_audit(args, state)
+        return
     srv = Server(codex_bin=args.codex_bin, codex_home=args.codex_home,
                  approval=args.approval,
                  progress_interval=args.progress_interval,
@@ -2969,9 +3622,15 @@ def main():
             # unittest，--no-verify 跳过，--verify 可指定命令。
             # 验测失败时 merge_task 已 reset 回合并前 HEAD，分支和
             # worktree 保留，任务标 merge-failed，不删改动。
+            # --no-verify 时 argv 一并跳过；两者都给时 argv 优先。
             verify_cmd = "" if args.no_verify else args.verify
+            verify_argv = None if args.no_verify else args.verify_argv
+            if verify_cmd and verify_argv:
+                print("verify_argv 已指定，--verify 字符串被忽略")
             ok, msg = merge_task(t["repo"], t["cwd"], t["branch"],
-                                 args.task, verify_command=verify_cmd)
+                                 args.task, verify_command=verify_cmd,
+                                 verify_argv=verify_argv,
+                                 audit_state_path=args.state)
             print(msg)
             if ok:
                 t["status"] = "merged"
@@ -3371,11 +4030,14 @@ def main():
                             args.approval_policy,
                             seed_dir=args.seed_dir)
             save_state(args.state, state)
+            if args.verify and args.verify_argv:
+                print("verify_argv 已指定，--verify 字符串被忽略")
             run_goal_loop(srv, state, args.state, task, args.task,
                           args.verify, args.max_rounds, args.model,
                           args.effort, args.sandbox, cfg=cfg,
                           timeout=args.turn_timeout,
-                          verify_timeout=args.verify_timeout)
+                          verify_timeout=args.verify_timeout,
+                          verify_argv=args.verify_argv)
         elif args.cmd == "goal-resume":
             # 断点丢失守卫（2026-10-29 长任务实弹）：host-state 里没有
             # 该任务时直接拒绝、正常退出，绝不能让 KeyError  traceback
@@ -3420,7 +4082,8 @@ def main():
                               start_round=1,
                               last_verify="",
                               last_turn_status=None,
-                              last_turn_text="")
+                              last_turn_text="",
+                              verify_argv=args.verify_argv)
                 return
             if t.get("status") not in ("goal-running", "interrupted",
                                        "running"):
@@ -3442,12 +4105,19 @@ def main():
             # P1：断点续跑时旧验收形态的 0 轮次（脏数据，0 在旧形态
             # 里表示不限轮次）不按不限轮次跑，钳制到 100 轮（与云端
             # GOAL_MAX_ROUNDS_CAP 同值）；新形态（verify 为空）的 0
-            # 保持不限轮次，上界是云端配额。
-            if max_rounds == 0 and t.get("verify_command"):
+            # 保持不限轮次，上界是云端配额。argv 形态同样是旧形态，
+            # 钳制条件须把 verify_argv 也算上，否则 argv 任务会被
+            # 当成新形态不限轮次跑。
+            if max_rounds == 0 and (t.get("verify_command")
+                                     or t.get("verify_argv")
+                                     or args.verify_argv):
                 max_rounds = 100
             srv.resume_thread(t["thread_id"])
             last_turn = (t.get("turns") or [{}])[-1]
             last_result = last_turn.get("result") or {}
+            resume_argv = (args.verify_argv
+                           if args.verify_argv is not None
+                           else t.get("verify_argv"))
             run_goal_loop(srv, state, args.state, t, t["goal"],
                           t["verify_command"], max_rounds,
                           args.model, args.effort, args.sandbox, cfg=cfg,
@@ -3457,7 +4127,8 @@ def main():
                           last_verify=t.get("goal_last_verify") or "",
                           last_turn_status=t.get("goal_last_turn_status"),
                           last_turn_text=last_result.get("text", "") or "",
-                          resume_note=args.resume_note)
+                          resume_note=args.resume_note,
+                          verify_argv=resume_argv)
         elif args.cmd == "parallel":
             with open(args.jobs) as f:
                 jobs = json.load(f)