给无人值守 AI Worker 配一张 JSON 检查卡

我几乎让一个 Worker 通宵跑了一整夜。CLI 还在和远程模型通信。我的笔记本盖子今晚已经合上一半了。到底是什么能证明这个 Worker 应该继续跑下去？

一个绿色的样本不算真正的关卡。我想要一个能够Fail Closed的文件。这个文件必须在我不在的时候也能正常工作。

周五的约束

这个 Worker 只做一个很窄的任务：把一个文件夹里的纯文本笔记分类，然后把JSON行写入磁盘。

今晚我不会守着这个循环跑。我不想为重试风暴买单。如果远程链路挂了，本地存根（stub）就接管。

时间上限四十五分钟。Token 上限只用免费模型。三次探测失败后就放弃，然后停止。

想要一个可以复制使用的工件而不是感觉？用一个 JSON 关卡文件加一个小的运行器。这两个文件就是整个启动仪式。

这个检查清单不是什么

这不是又一个模型横评。这不是 Agent 框架评测。这是一个 CLI 的"发货还是睡觉"检查清单。

如果一个关卡无法打印证据，就Fail。如果必需的关卡缺失，就不启动。运行器刻意保持Fail Closed状态。

我绝不跳过的八个关卡

我把这八行写在卡片上。然后复制到 gates.json。运行器是唯一可以说"走"的权威。

确认心跳路径可写。

把墙上时钟上限锁定为正整数。

把远程调用次数上限锁定为硬限制。

把空模型输出视为硬中止。

把输出文件限制在一个简短的允许列表内。

在下一个Tick时Honor STOP文件。

要求本地 stub 命令在 PATH 中。

磁盘剩余空间低于下限时不启动。

八个关卡对于一个任务来说足够了。第九个关卡就会变成宣言。下午六点我不需要宣言。

第一步——写关卡文件

把这个 JSON 保存到 Worker CLI 旁边。改任何值之前读两遍。注意这里根本没有"无限重试"这个 key。

{
  "task": "classify-notes",
  "heartbeat_path": "./run/heartbeat.txt",
  "max_wall_seconds": 3600,
  "max_remote_calls": 40,
  "empty_output_policy": "fail_closed",
  "output_allowlist": ["./out/", "./run/"],
  "stop_file": "./run/STOP",
  "stub_command": ["python", "stub_classify.py"],
  "min_free_mb": 512,
  "probe_attempts": 3,
  "probe_timeout_seconds": 8
}

缺失的那个 key 就是整个设计理念。通宵运行的 Worker 不应该自己发明新的重试策略。上限要写在文件里，而不是靠我记忆。

第二步——运行 Go/No-Go 检查器

运行器从不调用远程模型。它只回答启动还是等待。退出码 0 表示开始。码 2 表示等待。

#!/usr/bin/env python3
"""Unattended go/no-go gates. Copy, pin, and refuse defaults."""
from __future__ import annotations

import json
import shutil
import sys
import time
from pathlib import Path

def fail(msg: str) -> None:
    print(f"NO-GO: {msg}", file=sys.stderr)
    sys.exit(2)

def main() -> None:
    cfg = json.loads(Path("gates.json").read_text())
    Path("./run").mkdir(exist_ok=True)

    hb = Path(cfg["heartbeat_path"])
    try:
        hb.write_text(str(time.time()))
    except OSError as exc:
        fail(f"heartbeat not writable: {exc}")

    if int(cfg["max_wall_seconds"]) <= 0:
        fail("max_wall_seconds must be positive")
    if int(cfg["max_remote_calls"]) <= 0:
        fail("max_remote_calls must be positive")
    if cfg.get("empty_output_policy") != "fail_closed":
        fail("empty output must fail closed")

    stop = Path(cfg["stop_file"])
    if stop.exists():
        fail(f"stop file present: {stop}")

    for folder in cfg["output_allowlist"]:
        Path(folder).mkdir(parents=True, exist_ok=True)

    stub = cfg["stub_command"]
    if stub[0] not in {"python", "python3"} and shutil.which(stub[0]) is None:
        fail(f"stub command missing: {stub[0]}")

    free_mb = shutil.disk_usage(".").free // (1024 * 1024)
    if free_mb < int(cfg["min_free_mb"]):
        fail(f"disk free {free_mb}MB below floor")

    print("GO")
    sys.exit(0)

if __name__ == "__main__":
    main()

这个脚本是华丽的自动化吗？差得远。它能阻止一个糟糕的无人值守之夜吗？能的。离开键盘的时候我想要无聊的软件。

每次无人值守启动前这样运行它：

python go_nogo.py && python worker.py
echo $?

如果看到 NO-GO，就不启动。也不要说"就这一次"。这句话是小磁盘填满的途径。

第三步——保持一个失败fixture

我想要一个必须Fail Closed的fixture。一句博客句子不是证据。零字节的批处理文件才是证据。

mkdir -p fixtures out run
: > fixtures/empty_batch.jsonl
python - <<'PY'
from pathlib import Path
p = Path("fixtures/empty_batch.jsonl")
if p.stat().st_size == 0:
    raise SystemExit("FAIL-CLOSED: empty batch")
print("unexpected non-empty fixture")
PY

实际的 Worker 应该匹配那个 fixture。没有记录。不要用零行假扮成功。实时调用后零行是一起事故。

在同一规则内在写入器中接入：

def write_batch(path: str, rows: list[dict]) -> None:
    if not rows:
        raise RuntimeError("empty output: fail closed")
    target = Path(path)
    if not str(target).startswith(("out/", "run/", "./out/", "./run/")):
        raise RuntimeError("path outside allowlist")
    target.write_text("\n".join(json.dumps(r) for r in rows) + "\n")

你会发货一个庆祝沉默的分类器吗？我不会在今晚发那个分类器。沉默是整夜笔记消失的方式。

第四步——探测远程链路，然后退出

关卡文件把 probe_attempts 设为 3。那个数字就是放弃标准。在错误变得更好听之前不要提高它。

import subprocess
import sys

def fail(msg: str) -> None:
    print(f"NO-GO: {msg}", file=sys.stderr)
    sys.exit(2)

def probe_remote(cmd: list[str], attempts: int, timeout: int) -> None:
    last = None
    for _ in range(attempts):
        try:
            subprocess.run(cmd, check=True, timeout=timeout)
            return
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
            last = exc
    fail(f"remote probe failed {attempts} times: {last}")

只有在 go_nogo.py 退出码为 0 之后才调用探测。不要在一个半配置好的目录树里探测。顺序是检查清单的一部分，不是脚注。

一个免费的远程机器有助于这一步。我不想让我的笔记本当通宵宿主机。

披露：本文是 MonkeyCode 产品推广的一部分。

MonkeyCode 是一个开源项目，提供免费模型访问。它也为小规模运行提供免费服务器选项。我把这一对当作这些关卡的试验台，而且它是重放同一个 Go/No-Go 文件的一个地方。

如果三次探测都失败了，就运行 stub_classify.py。Stub 使用本地启发式规则，没有别的。它绝不会假装是远程模型。

# stub_classify.py — local, boring, honest
import json
import sys
from pathlib import Path

for line in Path(sys.argv[1]).read_text().splitlines():
    note = line.strip()
    if not note:
        continue
    label = "todo" if "todo" in note.lower() else "other"
    print(json.dumps({"text": note, "label": label, "source": "stub"}))

无聊的输出是特性，不是 bug。通宵 Worker 应该保持易于解释。如果我无法解释一个 Tick，我就不应该启动。

第五步——Tick 时带上 kill switch

刻意让无人值守循环保持短。每次 Tick 必须刷新心跳文件。每次 Tick 必须在任何调用之前 Honor STOP。

import time
from pathlib import Path

def run_ticks(max_calls: int, wall: int, stop: Path, hb: Path) -> None:
    start = time.time()
    calls = 0
    while calls < max_calls:
        if stop.exists():
            raise SystemExit("STOP file: abort")
        if time.time() - start > wall:
            raise SystemExit("wall clock: abort")
        hb.write_text(str(time.time()))
        # one remote or stub call lives here
        calls += 1
        time.sleep(1)

为什么用 STOP 文件而不是 Ctrl-C？因为我不会坐在那台键盘前。SSH 会话会断开，记忆不是方案。

从任何其他 Shell 中止可以用这个：

touch ./run/STOP

下一个 Tick 应该无需争论地死去。如果循环忽略 STOP，那是 bug。提 issue。不要把 bug 调教成重试。

时间、成本与回滚

在四十五分钟内把文件复制进去。第一个晚上不花任何付费 Token。要么一小时后停、要么四十次远程调用后停。

回滚就是 gates.json 里的 stub 命令。如果远程链路没了，就运行那个 stub。如果 stub 本身就没了，就不启动。

当以下任一情况发生时放弃整个 Worker：

三次远程探测连续失败

心跳文件在磁盘上变旧

输出出现在允许列表之外

可用磁盘降到 512MB 以下

最后那个关卡在小机器上坑过我。日志在卷满之前感觉是免费的。检查清单的存在就是为了让那个失败变得响亮。

谁不应该复制这个

如果你跑多租户 mesh，跳过这个。如果你需要正式的 SRE 覆盖，跳过这个。当任务不是一次性的时候跳过往。

这个检查清单是为独立 builder 准备的。它适配一个单任务 CLI 和一个短暂的夜晚。它不能替代真正的平台团队。

不要把 Worker 指向 secrets。不要把它指向生产写操作。不要让它在 out/ 和 run/ 之外创建文件。

我仍在持有的限制

我今天有空输出的关卡。我没有可信垃圾的关卡。这两种失败不是同一个 bug。

远程模型可能返回自信的胡说八道。JSON 可能解析零错误。标签可能完全错误。

那是我想要的下一个 Fail-Closed 检查。我不想要更大的 Agent 框架。我想要在文件里再加一个无聊的关卡。

如果你在免费远程机器上试这个，从空 fixture 开始。然后在一个 Tick 还在运行时 touch STOP。然后在临时文件夹里强制触发磁盘下限。

Worker 在每一个上都停了吗？如果它继续跑了，那检查清单骗了你。

在你的机器上，哪个无人值守关卡最先失败：空输出、心跳变旧，还是循环忽略的 STOP 文件？

要采取进一步行动，你可以考虑屏蔽这个人或举报滥用。