夜のうちに走らせたエージェントが、朝になると「成功5件・失敗0件」と報告してくる。ところが実際にサイトを開くと、3サイト分の更新がどこにも反映されていない。私自身、個人開発で4つのブログを毎日オフピークに分散して自動更新していますが、この「静かな失敗」に何度も足をすくわれてきました。
失敗そのものより怖いのは、失敗が記録されないまま消えることです。スケジュール実行のエージェントは、誰も見ていない時間帯に動きます。例外を握りつぶして 0 件と返した瞬間、その仕事は存在しなかったことになります。デッドレターは、この「なかったことにされる失敗」を必ず一箇所に集めるための仕組みです。
なぜ「失敗0件」が嘘になるのか
最も多い原因は、ループの中で例外を try/except で飲み込み、ログにも残さずに次へ進む書き方です。Antigravity のエージェントをスクリプトから起動する場合も同じ罠が待っています。タイムアウト、レート制限、トークン期限切れ、ネットワークの瞬断——どれも一過性に見えるため、つい「次回拾えばいい」と素通りしてしまいます。
ところが次回も同じ条件なら、同じ場所で同じように落ちます。落ちたジョブはキューから消え、再投入の手がかりも残りません。私の場合、AdMob のレポート取得を任せたジョブが週末の3日間ずっと無言で落ちていて、月曜にダッシュボードの数字が飛んでいて初めて気づいた、という経験があります。
設計の出発点は単純です。失敗は捨てるのではなく、退避させる 。これだけで運用の見通しが大きく変わります。
デッドレターに「何を」残すか
退避先に最低限残すべき情報は、再投入したときに同じ仕事を再現できるだけのコンテキストです。エラーメッセージだけでは足りません。次の項目を1行1レコードの JSON Lines で追記していくと、後から grep や jq で扱いやすくなります。
フィールド 役割
job_id 同じ仕事を一意に識別する（再投入の重複防止に使う） payload ジョブを再現するための入力一式 error_class 例外の型。再投入の可否を分岐させる鍵になる attempt これまでの試行回数 failed_at 失敗時刻（UTCで統一して保存する）
実装は驚くほど短く済みます。append-only にしておくと、複数のエージェントが同時に書いても行が壊れにくく、運用が楽になります。
import json, os, datetime, fcntl
DLQ_PATH = os.path.expanduser( "~/agent/dead_letter.jsonl" )
def to_dead_letter (job_id: str , payload: dict , exc: Exception , attempt: int ) -> None : record = { "job_id" : job_id, "payload" : payload, "error_class" : type (exc). __name__ , "message" : str (exc)[: 500 ], "attempt" : attempt, "failed_at" : datetime.datetime.now(datetime.timezone.utc).isoformat(), } os.makedirs(os.path.dirname( DLQ_PATH ), exist_ok = True ) with open ( DLQ_PATH , "a" , encoding = "utf-8" ) as f: fcntl.flock(f, fcntl. LOCK_EX ) f.write(json.dumps(record, ensure_ascii = False ) + " \n " ) fcntl.flock(f, fcntl. LOCK_UN ) ここで error_class をきちんと残しておくと、後段の再投入で「これは待てば直る失敗か、人が見るべき失敗か」を機械的に振り分けられます。これが次の階層設計の土台になります。
再投入を3つの階層に分ける
すべての失敗を一律にリトライすると、二つの事故が起きます。一過性でない失敗（例えば不正な入力）を延々と再試行して無駄なコストを払うか、逆に怖くなって一切再投入せず取りこぼすか、です。私は再投入を次の3階層に分けて運用しています。
即時リトライ（その場で2〜3回）
ネットワークの瞬断や 503 のような、明らかに一過性の失敗だけを対象にします。回数は2〜3回に厳しく制限し、間隔は次章のバックオフに従わせます。ここで粘りすぎないことが大切で、3回で直らない失敗は性質が違うと割り切ります。
遅延再投入（次のスケジュールで拾う）
即時リトライで直らなかったジョブを、デッドレターから次回の実行サイクルで拾い直します。レート制限（429）やトークン期限切れのように「時間を置けば直る」失敗がここに該当します。再投入の入口で job_id を照合し、すでに今サイクルで成功済みなら飛ばすようにしておくと、二重実行を防げます。
手動レビュー（人が判断する隔離枠）
入力スキーマの不一致、権限不足、想定外の例外型——こうした「待っても直らない」失敗は、再投入の対象から外して隔離します。attempt が上限（私は5回に設定しています）を超えたものも、同じ隔離枠へ送ります。ここに溜まったものだけを朝に確認すれば済むので、レビューの負担が劇的に軽くなります。
TRANSIENT = { "TimeoutError" , "ConnectionError" , "RateLimitError" } MAX_ATTEMPT = 5
def classify (record: dict ) -> str : if record[ "attempt" ] >= MAX_ATTEMPT : return "manual" if record[ "error_class" ] in TRANSIENT : return "retry" return "manual" 判断に迷ったら「待てば直るか」を基準にしてください。直らないものを retry に入れるとコストが膨らみ、直るものを manual に入れると人の手間が増えます。この振り分けの精度が、運用の静けさをそのまま決めます。
二重実行を防ぐ冪等キー
再投入で最も怖いのは、同じ仕事を二度実行してしまうことです。記事を二重投稿したり、課金処理を二回走らせたりすれば、デッドレターより大きな事故になります。job_id を冪等キーとして使い、成功したジョブの ID を別ファイルに記録しておきます。
DONE_PATH = os.path.expanduser( "~/agent/completed.jsonl" )
def already_done (job_id: str ) -> bool : if not os.path.exists( DONE_PATH ): return False with open ( DONE_PATH , encoding = "utf-8" ) as f: return any (job_id == json.loads(line)[ "job_id" ] for line in f if line.strip()) 再投入ループの先頭でこの照合を必ず通すようにすると、デッドレターに残ったまま実は前回成功していた、というケースで二重実行を避けられます。私はここを省いて記事を二重に push してしまい、後から片方を 410 で消す羽目になったことがあるので、強くお勧めします。
朝に5分で終わる確認手順
仕組みを作っても、見る習慣がなければ意味がありません。私は毎朝、隔離枠だけを次のコマンドで確認しています。集計を1行にしておくと、異常がない日は数秒で済みます。
# 当日の手動レビュー対象を error_class ごとに集計 jq -r 'select(.attempt >= 5 or (.error_class | test("Timeout|Connection|RateLimit") | not)) | .error_class' \ ~/agent/dead_letter.jsonl | sort | uniq -c | sort -rn このとき、同じ error_class が前日より急に増えていたら、それは個別ジョブの失敗ではなく構造的な変化のサインです。Antigravity 側のモデル更新や API の仕様変更が背景にあることも多く、私の運用ではこの「増分」を一次アラートとして扱っています。件数の絶対値より、前日との差分を見るほうが早く異変に気づけます。
仕組みとして残しておけば、失敗は怖い相手ではなくなります。退避させ、性質で振り分け、二重実行だけ防ぐ——この3つを通すだけで、夜間に動くエージェントを安心して任せられるようになります。次に着手するなら、まずは今動いているスケジュールジョブのうち1本に、この to_dead_letter を差し込むところから始めてみてください。同じ静かな失敗に悩んでいる方の助けになれば幸いです。