Antigravity のエージェントを「完了の自己申告」から卒業させる — 完了契約と外部検証の実装メモ

# scripts/done_check.py
import subprocess, sys, yaml, json
from pathlib import Path
 
EXIT_OK, EXIT_CHECK, EXIT_INTENT, EXIT_BUDGET = 0, 2, 3, 4
 
def sh(cmd: str) -> int:
    print(f"::group::{cmd}")
    rc = subprocess.run(cmd, shell=True).returncode
    print("::endgroup::")
    return rc
 
def changed_files() -> list[str]:
    out = subprocess.check_output(
        "git diff --name-only origin/main...HEAD", shell=True, text=True)
    return [l for l in out.splitlines() if l.strip()]
 
def main(contract_path: str) -> int:
    c = yaml.safe_load(Path(contract_path).read_text())
    task_dir = Path(contract_path).parent
 
    # 予算（変更範囲）を先に見る。逸脱は意図逸脱の強いシグナル
    files = changed_files()
    budget = c.get("budget", {})
    if (m := budget.get("max_files_changed")) and len(files) > m:
        print(f"BUDGET: {len(files)} files > {m}")
        return EXIT_BUDGET
    for forbidden in budget.get("forbid_paths", []):
        if any(f.startswith(forbidden) for f in files):
            print(f"BUDGET: touched forbidden path {forbidden}")
            return EXIT_BUDGET
 
    # 形式・機能。1つでも落ちたら即終了（後段の意図検証に進ませない）
    for layer, exit_code in (("formal", EXIT_CHECK), ("functional", EXIT_CHECK)):
        for cmd in c.get(layer, []):
            if sh(cmd) != 0:
                print(f"{layer.upper()} FAILED: {cmd}")
                return exit_code
 
    # 意図。質問への回答が「存在し・具体的か」を機械的にふるいにかける
    intent = c.get("intent", {})
    answers = (task_dir / "intent.md")
    if intent.get("questions"):
        if not answers.exists():
            print("INTENT: intent.md がありません")
            return EXIT_INTENT
        text = answers.read_text()
        for q in intent["questions"]:
            qid = q["id"]
            block = extract_answer(text, qid)
            if not block or len(block.strip()) < 40 or "確認しました" == block.strip():
                print(f"INTENT: 回答が不十分: {qid}")
                return EXIT_INTENT
            # 参照先（ファイル/テスト名）が含まれているか
            if "side_effects" in qid or "error_paths" in qid:
                if not any(tok in block for tok in (".ts", ".py", "test", "spec")):
                    print(f"INTENT: 参照先の明示なし: {qid}")
                    return EXIT_INTENT
 
    print("DONE: all layers passed")
    return EXIT_OK
 
def extract_answer(md: str, qid: str) -> str:
    # "## <qid>" 見出し直下のブロックを取り出す簡易パーサ
    lines, buf, capture = md.splitlines(), [], False
    for ln in lines:
        if ln.startswith("## "):
            capture = (qid in ln)
            continue
        if capture:
            buf.append(ln)
    return "\n".join(buf)
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1]))

# scripts/unattended_gate.py（無人ランの最終関門・抜粋）
import subprocess, sys, json
from pathlib import Path
 
STATE = Path(".agent/attempts.json")
 
def load_state():
    return json.loads(STATE.read_text()) if STATE.exists() else {}
 
def main(task_id: str, contract: str) -> int:
    st = load_state()
    rc = subprocess.run(["python", "scripts/done_check.py", contract]).returncode
    if rc == 0:
        # 差分監査：契約に列挙された forbid_paths 以外でも、巨大削除を検出
        deleted = subprocess.check_output(
            "git diff --numstat origin/main...HEAD", shell=True, text=True)
        big_del = [l for l in deleted.splitlines()
                   if l and l.split('\t')[1].isdigit() and int(l.split('\t')[1]) > 400]
        if big_del:
            print("AUDIT: 400行超の削除を検出。隔離します")
            return 5
        st[task_id] = 0
        STATE.write_text(json.dumps(st)); return 0
 
    st[task_id] = st.get(task_id, 0) + 1
    STATE.write_text(json.dumps(st))
    if rc in (4,) or st[task_id] >= 3:    # 予算逸脱 or 連続失敗で打ち切り
        print(f"QUARANTINE: rc={rc} attempts={st[task_id]}")
        return 9                          # 自動マージせず人間に回す
    print(f"RETRY: rc={rc} attempts={st[task_id]}")
    return rc
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1], sys.argv[2]))