Antigravity の長時間バッチエージェントを完走させる — 4層の耐久設計と私の実装メモ

第 1 層: チェックポイント粒度を 3 段階に分けて考える

長時間バッチを完走させる最初の手は、チェックポイントの粒度設計です。私は「タスク粒度」「ステップ粒度」「トークン粒度」の 3 段階で考えるようにしています。それぞれ性質が異なるため、組み合わせて使う前提です。

Step 1: タスク粒度のチェックポイント

タスク粒度とは「壁紙アプリ 30 本の AdMob 設定を見直す」のような、人間が見て意味のある一塊の単位です。1 タスク = 1 ファイルに分割し、進捗を JSON で保存します。

# checkpoint_task.py
import json, time, os
from pathlib import Path
 
CHECKPOINT_DIR = Path(".checkpoints/tasks")
CHECKPOINT_DIR.mkdir(parents=True, exist_ok=True)
 
def save_task(task_id: str, status: str, payload: dict) -> None:
    """タスク粒度のチェックポイントを原子的に書き込む。"""
    record = {
        "task_id": task_id,
        "status": status,           # pending / running / completed / failed
        "updated_at": time.time(),
        "payload": payload,
    }
    target = CHECKPOINT_DIR / f"{task_id}.json"
    tmp = target.with_suffix(".tmp")
    tmp.write_text(json.dumps(record, ensure_ascii=False, indent=2))
    os.replace(tmp, target)         # atomic rename

ポイントは os.replace による原子的なリネームです。書き込み途中にプロセスが落ちても、ファイルが半分書きの壊れた状態にはなりません。地味ですが、長時間バッチの耐久性は、こうした「落ちても壊れない継ぎ目」を一つずつ積み重ねた総和で決まると考えています。

Step 2: ステップ粒度のチェックポイント

ステップ粒度は、1 タスク内部の細かい工程です。たとえば「アプリ 1 本の処理 = 6 ステップ（取得→解析→提案→検証→適用→確認）」のような流れになります。タスクが落ちても、6 ステップ目から再開できれば、ほぼ無駄なリトライがなくなります。

私の運用では、ステップ粒度の保存は SQLite を使っています。理由は後述します。

Step 3: トークン粒度のチェックポイント

最も細かいのが、エージェントの内部状態（メッセージ列・要約・ツール呼び出し履歴）の保存です。これはステップ完了ごとではなく、一定トークン数（私は 8,000 トークン）ごとにスナップショットを取ります。コンテキスト爆発が起きそうな前に要約版に切り替えるための地点です。

「3 つも持つと管理が大変では？」と聞かれることもありますが、実際には粒度を分けたほうが圧倒的に楽です。タスク粒度だけだと再開時に 6 ステップ全部やり直し、ステップ粒度だけだとコンテキストが膨らみすぎる、という両極の苦しみを味わうことになります。

第 2 層: 永続化レイヤをどう選ぶか — SQLite / KV / Durable Objects / R2 の比較

チェックポイントを「どこに」保存するかは、想像以上に運用品質を左右します。私が試した 4 種類を比較します。

私が推奨するのは「開発時は SQLite、本番では KV + Durable Objects のハイブリッド」です。タスク粒度（個人で運営する各アプリの設定変更履歴など）は KV に、ステップ粒度の整合性が要る部分（AdMob 設定の 1 回限り適用）は Durable Objects に分けます。R2 は要約済みコンテキストの長期アーカイブで、Day 1 のスナップショットを後から振り返るために使っています。

実装の最小例として、SQLite 版を載せます。

# step_store.py
import sqlite3, json, time
from pathlib import Path
 
DB = Path(".checkpoints/steps.sqlite")
DB.parent.mkdir(parents=True, exist_ok=True)
 
def init_db() -> None:
    with sqlite3.connect(DB) as c:
        c.execute("""
            CREATE TABLE IF NOT EXISTS steps (
                task_id TEXT NOT NULL,
                step_idx INTEGER NOT NULL,
                status TEXT NOT NULL,
                payload TEXT NOT NULL,
                updated_at REAL NOT NULL,
                PRIMARY KEY (task_id, step_idx)
            )
        """)
 
def upsert_step(task_id: str, step_idx: int, status: str, payload: dict) -> None:
    with sqlite3.connect(DB) as c:
        c.execute(
            "INSERT OR REPLACE INTO steps VALUES (?, ?, ?, ?, ?)",
            (task_id, step_idx, status, json.dumps(payload, ensure_ascii=False), time.time()),
        )
 
def next_step(task_id: str) -> int:
    """次に実行すべきステップ番号を返す。完了済みは飛ばす。"""
    with sqlite3.connect(DB) as c:
        row = c.execute(
            "SELECT MAX(step_idx) FROM steps WHERE task_id = ? AND status = 'completed'",
            (task_id,),
        ).fetchone()
    return (row[0] + 1) if row and row[0] is not None else 0

INSERT OR REPLACE で冪等な上書きを保証し、MAX(step_idx) で再開点を計算します。SQLite はファイル 1 個で完結するので、Antigravity の VM が落ちても、外部ストレージに同期しておけば次回の VM で即座に復元できます。私はこの SQLite を Dropbox に置いて開発しているうちに、別 Mac の Antigravity でもそのまま再開できることに気づきました。朝に MacBook で書きはじめたバッチが、夕方に iMac で続きを走るのを見たとき、状態を外部に逃がしておくことの効きめを実感しました。

第 3 層: 冪等リストア — 同じ副作用を二度起こさない

チェックポイントを正しく取れても、リストア時に副作用が二重に起きると、本番では致命傷になります。AdMob のキャンペーン作成 API を 2 回叩いて重複作成、Stripe の課金を 2 回走らせる、メール通知が 2 通飛ぶ、いずれも私は経験があります。

冪等性を担保する原則は、外部副作用の前に「実行 ID」を生成し、KV に書き込む先回り方式です。

# idempotent_call.py
import hashlib, json, os
from typing import Callable, Any
 
def idempotent_call(
    operation_key: str,
    fn: Callable[[], Any],
    kv_get: Callable[[str], str | None],
    kv_put: Callable[[str, str], None],
) -> Any:
    """副作用付きの呼び出しを冪等化する。
 
    operation_key: 例 "admob/app123/campaign/2026-05-24"
    fn:            実際の副作用呼び出し
    kv_get/kv_put: KV ストアのアクセサ
    """
    cached = kv_get(operation_key)
    if cached is not None:
        return json.loads(cached)         # 二度目は実行せず結果を返す
 
    result = fn()
    kv_put(operation_key, json.dumps(result, ensure_ascii=False))
    return result
 
def make_key(*parts: str) -> str:
    """安定したキーを生成する。空白・記号は SHA1 で吸収。"""
    raw = "/".join(parts)
    return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16] + "_" + parts[0]

注意点として、kv_get → fn() → kv_put の間に他プロセスが同じキーで fn() を呼ぶ可能性があります。私は本番では Durable Objects の blockConcurrencyWhile で囲うか、KV に nx（既存なら失敗）相当のフラグを噛ませて、レースを潰しています。整合性が本気で要る箇所は Durable Objects、結果整合で許される箇所は KV、と使い分けるのがお勧めです。

副作用ごとに operation_key の設計をしておくと、後から監査ログにもなります。AdMob で何月何日にどのキャンペーンを触ったかを KV から逆引きできるので、私は月末の収益確認のときにも参照しています。

第 4 層: コンテキスト要約戦略 — 8,000 トークンごとに「過去」を縮める

長時間バッチで最も静かに壊れていくのが、エージェントのコンテキスト品質です。私の環境では、40,000 トークンを超えたあたりから提案の質が目に見えて落ちはじめ、80,000 トークンに近づくとループに陥ることが増えました。これは「過去のステップを覚えすぎている」のではなく、「過去の細部に引きずられて新しい判断ができない」状態です。

要約戦略はシンプルにします。8,000 トークン到達ごとに、過去のメッセージ列を「進捗サマリ + 直近 2,000 トークン」に圧縮し、それ以前は捨てます。捨てる、と言っても完全に消えるわけではなく、R2 にスナップショットを残します。

# context_summarizer.py
from dataclasses import dataclass
from typing import Callable
 
@dataclass
class Message:
    role: str       # user / assistant / tool
    content: str
    tokens: int
 
def summarize_context(
    messages: list[Message],
    summarize_fn: Callable[[list[Message]], str],
    threshold: int = 8000,
    keep_tail_tokens: int = 2000,
) -> list[Message]:
    total = sum(m.tokens for m in messages)
    if total < threshold:
        return messages
 
    # 末尾 keep_tail_tokens は維持
    tail: list[Message] = []
    used = 0
    for m in reversed(messages):
        if used + m.tokens > keep_tail_tokens:
            break
        tail.insert(0, m)
        used += m.tokens
 
    head = messages[: len(messages) - len(tail)]
    summary_text = summarize_fn(head)
    summary_msg = Message(role="assistant", content=f"[要約] {summary_text}", tokens=len(summary_text) // 2)
    return [summary_msg] + tail

実体験で言うと、要約戦略を入れる前の私のバッチは 12 時間で平均 $48 かかっていました。入れた後は同じワークロードで $14 まで下がりました。約 70% の削減です。質も落ちるどころか、要約のおかげで「いま何をやっているか」がエージェント自身にも明確になり、暴走が減りました。コンテキスト爆発は性能だけでなく、思考の純度の問題でもあるのだと感じています。

観測性 — チェックポイントが効いているかを 4 つの指標で測る

設計を入れたら、効いているかを必ず計測します。私が見ている指標は次の 4 つです。

• リスタート成功率: 意図的にプロセスを SIGKILL したとき、最後のステップから自動再開できる割合。目標 99%。
• 冗長実行率: 同じ operation_key が 2 回以上 fn() を呼んだ件数 / 総 operation_key 数。目標 0.1% 未満。
• コンテキスト圧縮率: 圧縮前トークン総量 / 圧縮後トークン総量。私の現状で約 4.2 倍。
• 平均バッチ完走時間: 同じワークロードでの完走時間の中央値。改善前後で比較する基準。

これらをスケジューラ側で 1 日 1 回集計し、Slack に投げています。観測指標がないと、せっかく組んだ耐久層がいつの間にか壊れていても気づきません。私自身、KV の TTL を間違えて 24 時間で消えるようにしてしまい、リスタート成功率が静かに 70% まで落ちていた事故があります。気づいたのは指標を見ていたからで、感覚だけだと半年放置しても不思議ではありませんでした。

コスト試算 — 24 時間バッチで実際に何円かかるか

ここまで来たら、コストの感覚を持っておくと意思決定が楽になります。私が個人で運営する壁紙アプリの AdMob 設定見直しを 24 時間バッチで走らせたときの実数値を共有します。

• 入力トークン: 約 1,800 万トークン（要約戦略あり）
• 出力トークン: 約 250 万トークン
• ツール呼び出し: 約 6,400 回（うち外部 API は 1,200 回）
• 推定総額: 約 ¥2,100（要約戦略なしだと約 ¥7,000）
• ストレージ: KV + Durable Objects + R2 で 月額 ¥80 程度

24 時間で約 ¥2,100、月にすると約 ¥63,000 になります。AdMob からの月次収益と比較すると、十分にペイする水準でした。とはいえ「ペイするから垂れ流していい」ではなく、要約戦略・冪等リストアの 2 層を抜いた瞬間に簡単に 3 倍以上に膨らみます。耐久設計は信頼性のためだけでなく、コスト管理の観点でも要になっています。

私がたどり着いた現実的な落としどころ

理想を全部入れると複雑になるので、私が個人開発の現場で落としどころにしている判断を 3 つ挙げます。

1. 30 分以内のバッチには、4 層全部は要らない。タスク粒度のチェックポイントだけで十分です。冪等性は本番副作用がある部分にだけ部分的に入れます。
2. 2 時間以上のバッチには、要約戦略を必ず入れる。コスト削減効果が大きく、思考品質も上がるからです。私は 60 分超えそうな時点でフラグを立てています。
3. 8 時間以上のバッチには、Durable Objects を採用する。KV だけだと結果整合の罠で「2 回実行してしまった」事故が起きやすいので、整合性が要る部分は強整合に倒します。

「最初から完璧に作ろう」とすると挫折します。私自身、4 層全てを最初から組んだわけではなく、半年かけて壊れ方を観察しながら、必要な層を 1 つずつ足していきました。耐久設計は、先に全体像を描いてから作るより、壊れた箇所を一つずつ補強していくうちに後から構造が立ち上がってくる——その順番のほうが、私には合っていると感じています。

次に試したいこと

ここまでが現時点での私の運用ですが、まだ手を入れたい領域がいくつかあります。Antigravity の Manager Surface 機能と組み合わせた「複数エージェントのチェックポイント共有」は、まだ実装が荒く、Day 1 のエージェントが Day 2 のエージェントに状態を引き渡すパターンを設計中です。それから、要約戦略の質を測る客観指標（要約前後で同じ判断を再現できる率）を持ち込みたいと考えています。

長時間バッチエージェントは、個人開発者が片手間で運営しているメディアやアプリ事業を、もう一段階上のスケールに連れていける道具だと感じています。私自身、まだ学びの途中ですが、同じ課題に取り組んでいる方の参考になれば嬉しいです。お読みいただき、ありがとうございました。