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

dev.to · AI#AI Worker#工程化#可靠性

Editor brief · 编辑速览

{ "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 }

#!/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()

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

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")

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}")

# 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"}))

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)

完整中文译文