◈ Agents & Manager/2026-06-13上級

Antigravity SDK で定期実行エージェントを冪等に組む — 二重起動と取りこぼしを防ぐ実装

Antigravity SDK のスケジュール実行は、cron 式を一行足すだけで動きます。難しいのはそこではなく、同じジョブが二重に走ったり、前回の続きを取りこぼしたりせず「何度動いても結果が一つ」になる実装です。個人開発の定期メンテナンスで踏んだ失敗をもとに、冪等な定期エージェントの組み方をまとめました。

agents⁶⁷ antigravity-sdk scheduled-execution³ idempotency³ automation²⁴

✦ プレミアム記事

定期実行のエージェントを初めて SDK で組んだとき、最初の一週間は順調でした。問題が見えたのは、デプロイの最中に手動でもう一度トリガーを引いた朝です。同じレポート集計のジョブが二つ同時に走り、片方が書き込んだファイルをもう片方が上書きして、その日の数値だけが前日のコピーになっていました。

cron 式を足してエージェントを定期起動させること自体は、Antigravity SDK ではほとんど一行で済みます。本当に難しいのは、起動が一回だろうと二回だろうと、結果が必ず一つに収束する状態をつくることです。

個人開発で複数のアプリを運用していると、依存更新・クラッシュレポートのトリアージ・AdMob のレポート集計といった作業を毎晩エージェントに任せたくなります。そのとき土台になるのが、この「冪等性」でした。

スケジュール実行を定義する最小構成

まず、SDK でスケジュール実行を宣言する最小の形を見ておきます。ポイントは、エージェント本体（何をするか）とスケジュール定義（いつ動かすか）を分けて書けることです。

# schedule_agent.py
from antigravity import Agent, Schedule, run
 
agent = Agent(
    name="daily-report-aggregator",
    model="gemini-3.5-flash",
    instructions="""
    指定された日付の AdMob レポートを集計し、
    output/report-<date>.json に書き出してください。
    既存ファイルがあれば上書きせず終了してください。
    """,
)
 
schedule = Schedule(
    agent=agent,
    # 毎日 JST 09:00（UTC 00:00）
    cron="0 0 * * *",
    timezone="Asia/Tokyo",
    # 1回の実行に上限を設け、暴走を物理的に止める
    max_duration_seconds=600,
)
 
if __name__ == "__main__":
    run(schedule)

cron と timezone を指定すれば、スケジューラ側が時刻になるたびにエージェントを起動します。max_duration_seconds は地味ですが重要で、想定外のループに入ったジョブを時間で強制終了させる安全弁になります。

ここまでは公式の例にも近い形です。事故が起きるのは、この先の「起動のされ方」を誤解したときでした。

「毎回まっさらなセッション」という前提を取り違えない

Antigravity のスケジュール実行は、起動のたびに新しいセッションを作ります。前回の会話履歴も、メモリ上の変数も引き継ぎません。これは暴走の伝播を防ぐ良い設計なのですが、裏を返すと「前回どこまで終わったか」をエージェント自身は知らない、ということです。

最初に私が書いたコードは、暗黙にセッションが継続する前提になっていました。「昨日の続きから集計」といった指示を書いていたのですが、新規セッションには昨日の状態がないので、エージェントは毎回ゼロから全期間を集計し直していました。結果は合っていたものの、実行時間が日々伸びていきます。

正しくは、状態を外部に置き、起動のたびに読み直す設計にします。

import json, pathlib
 
STATE = pathlib.Path("state/last_processed.json")
 
def load_checkpoint() -> str | None:
    if STATE.exists():
        return json.loads(STATE.read_text())["last_date"]
    return None
 
def save_checkpoint(date: str) -> None:
    STATE.parent.mkdir(parents=True, exist_ok=True)
    STATE.write_text(json.dumps({"last_date": date}))

セッションは毎回まっさらでも、チェックポイントだけはディスク（や KV）に残ります。エージェントの指示文には「state/last_processed.json を読み、その翌日から処理する」と書いておけば、新規セッションでも続きから動けます。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦Antigravity SDK でスケジュール実行エージェントを定義する最小コードと、cron 起動・新規セッション生成・状態の引き継ぎがどう噛み合うかを把握できる

✦二重起動・取りこぼし・途中失敗の3つの典型事故に対して、ロック・チェックポイント・構造化ログで冪等性を担保する実装をそのまま流用できる

✦ドライラン検証と pause/resume の運用フックを組み込み、定期ジョブを10分でレビューできる状態に持っていける

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

冪等性を担保する3つのガード

二重起動・取りこぼし・途中失敗。定期ジョブで踏む事故はだいたいこの3つに収まります。それぞれに対応するガードを順番に入れていきます。

1. 実行ロックで二重起動を止める

冒頭の事故の直接の原因はこれでした。スケジュール起動と手動トリガーが重なると、同じジョブが並走します。最小限のファイルロックで弾きます。

import os, time, contextlib
 
LOCK = pathlib.Path("state/agent.lock")
 
@contextlib.contextmanager
def single_run(stale_seconds: int = 1800):
    # 古いロックは残骸とみなして無視する
    if LOCK.exists() and time.time() - LOCK.stat().st_mtime < stale_seconds:
        raise RuntimeError("already running; skip this trigger")
    LOCK.write_text(str(os.getpid()))
    try:
        yield
    finally:
        LOCK.unlink(missing_ok=True)

stale_seconds を設けているのは、プロセスが異常終了してロックが残った場合に、永遠に起動できなくなるのを防ぐためです。前回のクラッシュでロックファイルだけ残る、というのは実運用で必ず一度は起きます。

2. 出力キーで取りこぼしと重複書き込みを防ぐ

冪等性のいちばん効く考え方は「結果を、入力から決まる一意のキーに紐づける」ことです。日付ごとのレポートなら、ファイル名そのものを出力キーにします。

def already_done(date: str) -> bool:
    return pathlib.Path(f"output/report-{date}.json").exists()
 
def process(date: str):
    if already_done(date):
        return "skip"          # 何度走っても二度処理しない
    data = aggregate_admob(date)
    pathlib.Path(f"output/report-{date}.json").write_text(json.dumps(data))
    save_checkpoint(date)
    return "done"

この形にしておくと、同じ日に何度起動されても結果は一つに収束します。「実行回数」ではなく「処理対象の集合」で正しさを定義できるのが、冪等な設計の利点です。

3. チェックポイントは処理の後に進める

順番を一つ間違えると、途中失敗で取りこぼしが起きます。チェックポイントを「処理の前」に進めてしまうと、その直後に落ちたとき、未処理の日付が「済み」と記録されて二度と拾われません。

必ず、出力の書き込みが成功してからチェックポイントを進めます。先ほどの process() で save_checkpoint(date) を write_text の後に置いているのは、このためです。地味ですが、ここの順序がすべてです。

失敗を「無音」にしない構造化ログ

定期ジョブの一番こわい失敗は、エラーで止まることではなく、何事もなかったかのように「成功」を出し続けることです。集計対象がゼロ件でも、書き込みに失敗していても、ログが OK だけだと気づけません。

実行のたびに、判断材料になる数値を構造化して残します。

import datetime, sys, json
 
def log_run(status: str, **fields):
    record = {
        "ts": datetime.datetime.now(datetime.UTC).isoformat(),
        "agent": "daily-report-aggregator",
        "status": status,
        **fields,
    }
    print(json.dumps(record, ensure_ascii=False), file=sys.stderr)
 
# 使用例
log_run("done", date="2026-06-13", rows=412, bytes_written=18044)
log_run("skip", date="2026-06-13", reason="already_done")
log_run("empty", date="2026-06-13", rows=0)  # ← これが拾えることが大事

rows=0 を独立した状態として出すのがコツです。「成功したが中身が空」を done に混ぜないことで、週次レビューのときに status:empty を検索するだけで異常が浮かびます。私自身、この一手間を入れてから、無音で空回りしていたジョブを二件見つけました。

ドライランで起動前に挙動を確かめる

スケジュール実行は、間違っていても深夜に静かに動くので、デプロイ前の検証が効きにくい領域です。SDK のローカル実行を使い、cron を待たずに一回ぶんを手元で流せるようにしておきます。

# 特定日付を渡して一回だけ実行（cron を待たない）
python schedule_agent.py --once --date 2026-06-13 --dry-run
 
# 冪等性の確認：同じコマンドを二回流して結果が一つに収束するか
python schedule_agent.py --once --date 2026-06-13
python schedule_agent.py --once --date 2026-06-13   # → skip になるはず

--dry-run では実際の書き込みをせず、log_run の出力だけを確認します。二回連続で流して二回目が skip になることを、デプロイ前の最低限のチェックにしています。ここが done, done になるなら、出力キーの設計がまだ甘いということです。

pause / resume を運用フックに組み込む

最後に、運用で必ず必要になるのが「一時的に止める」手段です。デプロイ中・障害調査中・上流 API のメンテ中など、止めたい場面は定期的に来ます。

PAUSE = pathlib.Path("state/PAUSED")
 
def guard_paused():
    if PAUSE.exists():
        log_run("paused", reason=PAUSE.read_text().strip() or "manual")
        sys.exit(0)   # 失敗ではなく正常終了として抜ける

state/PAUSED というファイルを置くだけで次回以降の起動をスキップし、消せば再開する、という単純な仕組みです。SDK の schedule.pause() / schedule.resume() をスクリプトから叩く方法もありますが、ファイル一つで止められるフックを併設しておくと、SDK の API が使えない緊急時にも手で止められて安心です。

ここまでをまとめると、冪等な定期エージェントの骨格は「ロックで並走を防ぐ → 出力キーで重複を弾く → 処理の後にチェックポイント → 状態を構造化ログに残す」という流れになります。

次に試すなら、いま動かしている定期ジョブを一つ選び、--once を二回連続で流してみてください。二回目が skip で終わらないなら、そのジョブはまだ二重起動に弱いということです。そこが、冪等化の最初の一歩になります。

本番運用に乗せる前に一つだけ注意点を添えるなら、ロックとチェックポイントを別々のストレージに置かないことです。私の場合、ロックをローカルファイル・状態を KV に分けた時期があり、KV 側の書き込みが遅延した隙に二重処理が起きました。同じ整合性で守りたいものは、同じ場所にまとめて置くことをお勧めします。

定期実行は手放しで回したくなる領域ですが、最初の一週間だけは毎朝ログを覗くことを習慣にすると、無音の事故をずっと早く拾えます。