feat(ack): add structured worker model routing
This commit is contained in:
@@ -1,171 +1,30 @@
|
||||
#!/usr/bin/env python3
|
||||
"""校验 Music Pilot 的 Developer/Test worker 启动命令。
|
||||
"""Fail-closed migration shim for ACK's removed free-form command validator.
|
||||
|
||||
用法：
|
||||
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 用法或命令解析错误。
|
||||
ACK v0.10 no longer accepts a shell command as worker configuration. Keep this
|
||||
file only so older documentation or automation fails with a precise migration
|
||||
message instead of silently retaining the unsafe path.
|
||||
"""
|
||||
|
||||
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
|
||||
MIGRATION_MESSAGE = (
|
||||
"ACK v0.10 已停用自由 worker command 校验器；请在 "
|
||||
"docs/ack/tasks.yaml 的 project.orchestration 中声明 profile，并仅调用 "
|
||||
"scripts/launch_worker.py profile-hash|plan|launch。"
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
arguments = list(sys.argv[1:] if argv is None else argv)
|
||||
if arguments == ["--self-test"]:
|
||||
print("legacy worker command 路径已 fail-closed")
|
||||
return 0
|
||||
sys.stderr.write(MIGRATION_MESSAGE + "\n")
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user