feat(ack): enforce worker launch policy

2026-07-12 17:19:46 +08:00
parent ae05845ee7
commit b3765ecad6
6 changed files with 239 additions and 22 deletions
@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+"""校验 Music Pilot 的 Developer/Test worker 启动命令。
+
+用法：
+  python3 validate_worker_command.py --role developer --command '<command>'
+  python3 validate_worker_command.py --role test --command '<command>'
+  python3 validate_worker_command.py --role developer --upgraded --command '<command>'
+  python3 validate_worker_command.py --self-test
+
+退出码：0 通过 / 1 规则不通过 / 2 用法或命令解析错误。
+"""
+
+from __future__ import annotations
+
+import argparse
+import shlex
+import sys
+from pathlib import Path
+
+
+CODEX_YOLO = "--dangerously-bypass-approvals-and-sandbox"
+CURSOR_YOLO = "--yolo"
+
+
+def option_value(tokens: list[str], *names: str) -> str | None:
+    for index, token in enumerate(tokens):
+        for name in names:
+            if token == name and index + 1 < len(tokens):
+                return tokens[index + 1]
+            prefix = f"{name}="
+            if token.startswith(prefix):
+                return token[len(prefix) :]
+    return None
+
+
+def codex_effort(tokens: list[str]) -> str | None:
+    configs: list[str] = []
+    for index, token in enumerate(tokens):
+        if token in {"-c", "--config"} and index + 1 < len(tokens):
+            configs.append(tokens[index + 1])
+        elif token.startswith("--config="):
+            configs.append(token.split("=", 1)[1])
+    for config in configs:
+        if config.startswith("model_reasoning_effort="):
+            return config.split("=", 1)[1].strip('"\'')
+    return None
+
+
+def validate(role: str, command: str, upgraded: bool = False) -> list[str]:
+    try:
+        tokens = shlex.split(command)
+    except ValueError as exc:
+        return [f"命令无法解析：{exc}"]
+    if not tokens:
+        return ["启动命令不能为空"]
+
+    executable = Path(tokens[0]).name
+    errors: list[str] = []
+
+    if upgraded and role != "developer":
+        errors.append("只有 Developer 可以使用 --upgraded")
+
+    if executable == "codex":
+        if CODEX_YOLO not in tokens:
+            errors.append(f"Codex worker 必须包含 {CODEX_YOLO}")
+
+        model = option_value(tokens, "-m", "--model")
+        effort = codex_effort(tokens)
+        if upgraded:
+            expected_model = "gpt-5.6-sol"
+            allowed_efforts = {"high", "xhigh"}
+        elif role == "developer":
+            expected_model = "gpt-5.6-terra"
+            allowed_efforts = {"medium"}
+        else:
+            expected_model = "gpt-5.6-luna"
+            allowed_efforts = {"low"}
+
+        if model != expected_model:
+            errors.append(
+                f"Codex {role} 模型应为 {expected_model}，实际为 {model or '未指定'}"
+            )
+        if effort not in allowed_efforts:
+            expected = "/".join(sorted(allowed_efforts))
+            errors.append(
+                f"Codex {role} reasoning effort 应为 {expected}，实际为 {effort or '未指定'}"
+            )
+    elif executable == "cursor-agent":
+        if upgraded:
+            errors.append("Cursor worker 不使用 Codex --upgraded 映射")
+        if CURSOR_YOLO not in tokens:
+            errors.append(f"Cursor worker 必须显式包含 {CURSOR_YOLO}")
+        model = option_value(tokens, "--model")
+        if model != "auto":
+            errors.append(f"Cursor {role} 模型应为 auto，实际为 {model or '未指定'}")
+    else:
+        errors.append(f"不支持的 worker CLI：{executable}；只允许 codex 或 cursor-agent")
+
+    return errors
+
+
+def run_self_test() -> int:
+    cases = [
+        (
+            "codex developer",
+            "developer",
+            f"codex {CODEX_YOLO} -m gpt-5.6-terra -c model_reasoning_effort=medium",
+            False,
+            True,
+        ),
+        (
+            "codex test",
+            "test",
+            f"codex {CODEX_YOLO} -m gpt-5.6-luna -c model_reasoning_effort=low",
+            False,
+            True,
+        ),
+        (
+            "codex upgraded developer",
+            "developer",
+            f"codex {CODEX_YOLO} -m gpt-5.6-sol -c model_reasoning_effort=high",
+            True,
+            True,
+        ),
+        ("cursor worker", "test", "cursor-agent --yolo --model auto", False, True),
+        ("naked codex", "developer", "codex", False, False),
+        (
+            "wrong codex role model",
+            "test",
+            f"codex {CODEX_YOLO} -m gpt-5.6-terra -c model_reasoning_effort=medium",
+            False,
+            False,
+        ),
+        ("cursor without yolo", "developer", "cursor-agent --model auto", False, False),
+    ]
+    failures: list[str] = []
+    for name, role, command, upgraded, expected_pass in cases:
+        passed = not validate(role, command, upgraded)
+        if passed != expected_pass:
+            failures.append(name)
+    if failures:
+        sys.stderr.write("worker 命令校验器自测失败：" + ", ".join(failures) + "\n")
+        return 1
+    print(f"worker 命令校验器自测通过：{len(cases)} 项")
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="校验 Developer/Test worker 启动命令")
+    parser.add_argument("--role", choices=("developer", "test"))
+    parser.add_argument("--command")
+    parser.add_argument("--upgraded", action="store_true", help="校验升级后的 Codex Developer")
+    parser.add_argument("--self-test", action="store_true")
+    args = parser.parse_args(argv)
+
+    if args.self_test:
+        return run_self_test()
+    if not args.role or not args.command:
+        parser.error("非自测模式必须同时提供 --role 和 --command")
+
+    errors = validate(args.role, args.command, args.upgraded)
+    if errors:
+        sys.stderr.write("worker 启动命令校验失败：\n")
+        for error in errors:
+            sys.stderr.write(f"  - {error}\n")
+        return 1
+    print(f"worker 启动命令校验通过：role={args.role}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())