移行を終えた翌朝、CI が赤いまま固定されていました。
前日までは、10回に4回ほどは通っていたテストです。それが移行後は一度も通らない。手を入れた覚えのある箇所は、フックの登録方法だけでした。
先に結論を書きます。これは移行の失敗ではありませんでした。むしろ、移行が効いたことの証拠でした。
変わったのは「何を登録できるか」ではなく「いつ決まるか」
Antigravity SDK では、セッションを初期化したあとに動的にフックとトリガーを登録する方式が廃され、セッション生成時の宣言へ寄せられました。版によって細部は動きますので、実際の記述は Antigravity の changelog でご確認いただければと思います。
登録できる内容そのものは変わっていません。変わったのは、構成が確定するタイミングです。
観点 初期化後の動的登録 セッション生成時の宣言
構成が確定する時点 実行中（登録が終わった順） セッション生成の瞬間 フックの実行順 登録完了の順に依存する 書いた配列の順 条件分岐の置き場所 登録するかどうかで表現できる フックの内部で表現する 同じ入力に対する再現性 実行ごとに変わりうる 常に同じ 構成の検査 実行してみないと分からない 起動前に静的に読める
私はこの表の最終行のために移行した、と思っていました。起動前に構成を読めることの価値は分かりやすいからです。実際に効いたのは、その1つ上の行でした。
移行対象を機械的に洗い出す
最初は grep -rn "register_hook" で足りると考えておりました。個人開発で持っているエージェント定義は4つだけです。目視でも追えるだろう、と。
足りませんでした。register_hook が条件分岐の中にあるのか、ループの中にあるのか、try の中にあるのかで、書き換えの手間がまったく違ったからです。grep は行を返しますが、その行が置かれている文脈は返してくれません。
そこで、呼び出し位置とその構文的な文脈をまとめて出す小さなスクリプトを書きました。
#!/usr/bin/env python3 """初期化後のフック/トリガー動的登録を洗い出し、移行の難易度で仕分ける。""" import ast, os, sys
DYNAMIC = { "register_hook" , "register_trigger" , "add_hook" , "add_trigger" }
def scan (root): rows = [] for dp, _, fs in os.walk(root): if any (seg in dp for seg in ( ".venv" , "node_modules" , "__pycache__" )): continue for f in fs: if not f.endswith( ".py" ): continue p = os.path.join(dp, f) try : tree = ast.parse( open (p, encoding = "utf-8" ).read(), p) except SyntaxError as e: # 移行途中のファイルが混ざると必ずここに来る。黙って飛ばさない print ( f "skip (syntax): { p } : { e } " , file = sys.stderr) continue
# ast には親参照がないため、先に子→親の対応表を作っておく parent = {} for n in ast.walk(tree): for c in ast.iter_child_nodes(n): parent[c] = n
for n in ast.walk(tree): if not ( isinstance (n, ast.Call) and isinstance (n.func, ast.Attribute)): continue if n.func.attr not in DYNAMIC : continue ctx, cur = set (), parent.get(n) while cur is not None : if isinstance (cur, ast.If): ctx.add( "conditional" ) elif isinstance (cur, (ast.For, ast.While, ast.comprehension)): ctx.add( "loop" ) elif isinstance (cur, (ast.Try, ast.ExceptHandler)): ctx.add( "try" ) cur = parent.get(cur) rows.append((p, n.lineno, n.func.attr, sorted (ctx))) # ast.walk は幅優先なのでソース順にならない。ここで必ず並べ直す return sorted (rows, key =lambda r: (r[ 0 ], r[ 1 ]))
TIER = { frozenset (): "A: そのまま配列へ移せる" , frozenset ({ "conditional" }): "B: フック内の早期 return へ書き換える" , frozenset ({ "loop" }): "B: 生成した関数を配列へ展開する" , frozenset ({ "try" }): "C: 失敗時の挙動を先に決める" , }
def main (root): rows = scan(root) for p, line, call, ctx in rows: tier = TIER .get( frozenset (ctx), "C: 手作業で判断する" ) print ( f " { p } : { line } { call :<17 } ctx= { ',' .join(ctx) or '-' :<12 } { tier } " ) print ( f " \n 合計 { len (rows) } 箇所" )
if __name__ == "__main__" : main(sys.argv[ 1 ] if len (sys.argv) > 1 else "." ) ast.walk が幅優先で走るため、そのまま出力するとソース順に並びません。移行作業はファイルを上から順に潰していくものですから、ここで並べ直しておかないと手戻りが出ます。最初にこれで一度混乱しました。
手元の定義に当てたときの出力です。
sample/agents/build_agent.py:4 register_hook ctx=- A: そのまま配列へ移せる
sample/agents/build_agent.py:6 register_hook ctx=conditional B: フック内の早期 return へ書き換える
sample/agents/build_agent.py:7 register_trigger ctx=- A: そのまま配列へ移せる
sample/agents/review_agent.py:5 register_hook ctx=loop B: 生成した関数を配列へ展開する
sample/agents/review_agent.py:6 register_trigger ctx=- A: そのまま配列へ移せる
合計 5 箇所
A が3件、B が2件。数としては小さいのですが、この仕分けがあると「今日はどこまで進むか」の見積もりが立ちます。C（try の中の登録）が出た場合だけは、書き換えの前に「登録に失敗したら止めるのか、続けるのか」を決める必要があります。
A のものは、配列へ移すだけで終わる
条件も繰り返しもない登録は、機械的な作業です。
# 移行前: セッションを作ってから足していく def build (session): session.register_hook( "pre_tool_use" , audit_hook) session.register_trigger( "on_file_write" , reindex) return session # 移行後: 生成時に渡し切る from google.antigravity import AgentConfig
config = AgentConfig( hooks = { "pre_tool_use" : [audit_hook]}, triggers = { "on_file_write" : [reindex]}, ) 差分としては地味です。ただ、この形になった時点で、構成を読むのに実行が要らなくなります。
B は、条件を「登録するか」から「実行するか」へ移す
ここで一度、遠回りをしました。
私は最初、条件を宣言側に持ち込もうとしました。フラグごとに AgentConfig を用意し、起動時にどれかを選ぶ形です。書き始めてすぐに手が止まりました。フラグが3つあれば8通りの構成が要ります。増えるたびに倍になる定義を、自分で保守できる気がしませんでした。
正しかったのは逆向きでした。宣言には静的に全部並べておき、実行するかどうかはフックの中で決める。
# 移行前: 条件が「登録するかどうか」に乗っている if feature_flags.get( "strict_diff" ): session.register_hook( "post_tool_use" , diff_guard) # 移行後: 条件はフックの中へ。宣言は常に同じ形になる def diff_guard (event, ctx): if not ctx.flags.get( "strict_diff" ): return # 無効時は何もしないフックとして存在し続ける ...
config = AgentConfig( hooks = { "post_tool_use" : [diff_guard]}) 宣言は静的なもの、判断は実行時のもの。この線引きを一度決めてしまうと、残りは迷わなくなりました。
ループも同じ考え方で畳めます。生成側を関数に切り出し、配列へ展開します。
# 移行前 for name in ( "lint" , "typecheck" , "test" ): session.register_hook( "post_tool_use" , make_gate(name)) # 移行後（並び順がここで目に見える形で固定される） GATES = [ "lint" , "typecheck" , "test" ] config = AgentConfig( hooks = { "post_tool_use" : [make_gate(n) for n in GATES ]}, ) 宣言の中に内包表記を置いてよいのかは迷いました。結果としては、置いてよいと考えています。評価されるのは生成の瞬間の一度きりで、実行時に構成が変わらないという前提は保たれるためです。
誰も並び順を決めていなかった
移行で一番時間を使ったのは、この一点でした。
登録が非同期の初期化処理に散っていたころ、フックの実行順は「そのとき先に登録が終わったもの順」でした。設定ファイルの読み込みや MCP サーバーへの接続が挟まれば、順序は実行ごとに変わります。
どの程度変わるのか、感覚では分からなかったので測りました。SDK そのものではなく、両方式で順序がどう決まるかだけを再現した最小構成です。
import asyncio, random, collections, time
HOOKS = [ "audit" , "diff_guard" , "reindex" , "metrics" , "cost_cap" ]
# 方式A: 初期化後に動的登録（各セットアップが非同期I/Oを挟む） async def dynamic_register (sink, name): await asyncio.sleep(random.uniform( 0.0 , 0.002 )) # 設定読込・MCP接続を模す sink.append(name)
async def run_dynamic (): sink = [] await asyncio.gather( * (dynamic_register(sink, h) for h in HOOKS )) return tuple (sink)
# 方式B: セッション生成時に宣言（配列の順がそのまま実行順） def run_declarative (): return tuple ( HOOKS )
N = 500 t0 = time.perf_counter() dyn = collections.Counter(asyncio.run(run_dynamic()) for _ in range (N)) t1 = time.perf_counter() dec = collections.Counter(run_declarative() for _ in range (N))
print ( f "動的登録: 観測された実行順の種類 = { len (dyn) } " ) for o, c in dyn.most_common( 3 ): print ( f " { c / N * 100 :5.1f } % { ' -> ' .join(o) } " ) print ( f "宣言的登録: 観測された実行順の種類 = { len (dec) } " ) 500回の実行で、出力はこうなりました。
動的登録: 観測された実行順の種類 = 119
1.8% audit -> diff_guard -> cost_cap -> reindex -> metrics
1.6% diff_guard -> cost_cap -> metrics -> audit -> reindex
1.6% cost_cap -> metrics -> reindex -> audit -> diff_guard
宣言的登録: 観測された実行順の種類 = 1
5つのフックの並べ方は120通りです。そのうち119通りが500回のあいだに出ていました。最も多い順序でも出現率は1.8%。特定の順序が優勢になることは、ありませんでした。
これだけなら「順序はばらつく」で終わります。問題は、フックのあいだに依存があるときです。私の構成では cost_cap が予算の上限を確定させ、diff_guard がそれを読んで判定していました。順序が逆になれば、判定は未確定の値を読みます。
async def dynamic_once (): sink = [] await asyncio.gather( * (reg(sink, h) for h in [ "audit" , "diff_guard" , "reindex" , "metrics" , "cost_cap" ])) return sink.index( "cost_cap" ) < sink.index( "diff_guard" ) # True = 依存が満たされた
N = 2000 ok = sum (asyncio.run(dynamic_once()) for _ in range (N)) print ( f "動的登録 { N } 回: 依存が満たされた割合 = { ok / N * 100 :.1f } %" ) 動的登録 2000 回: 依存が満たされた割合 = 48.2% (満たされない = 51.8%)
宣言的登録: 依存が満たされた割合 = 100.0%
2,000回で48.2%。ほぼコイン投げでした。
「10回に4回ほど通る」というテストの体感は、この数字とよく合います。落ちていたのは環境のせいでも、タイミングの微妙なずれのせいでもありませんでした。順序を誰も決めていなかった、それだけのことでした。
「毎回落ちる」は、直すべきものが見えている状態
冒頭のCIに戻ります。
宣言へ移した時点で、実行順は私が書いた配列のとおりに固定されます。そして最初に書いた配列は、たまたま依存を満たさない側でした。diff_guard を先に、cost_cap を後に並べていたのです。移行前の順序を思い出しながら書いたわけではなく、単にコードに現れる順で並べただけでした。
直感では、順序が固定されれば不安定なテストは安定して通るようになる、と思っていました。起きたのは逆でした。安定して落ちるようになったのです。
宣言的な構成は、不具合を消すものではありませんでした。不具合を再現可能にするものでした。
この違いは大きいと感じています。48.2%で落ちる不具合は、直したかどうかの確認ができません。2回通っても、それは偶然かもしれない。100%で落ちる不具合は、1回通れば直っています。失敗を手元へ確実に持ってくるという点では、Antigravity Agent の Record & Replay — 失敗を3分で再現する本番運用パターン で扱った考え方と同じ方向にあります。
移行直後にCIが赤で固定されたら、まず疑うべきは移行の失敗ではなく、宣言した並び順です。私はここで半日を溶かしました。同じ場所で止まる方がいらっしゃれば、配列を先に見ていただければと思います。
並び順を決めるときに置いた3つの基準
配列を書き直すにあたって、その場の判断で並べると同じことを繰り返すと考え、基準を先に決めました。
位置 置くもの 理由
先頭 上限・予算の確定（cost_cap 等） 後続の判定が読む値を、判定より前に確定させる 中盤 記録・監査（audit、metrics） 拒否されたものも含めて記録に残したい 末尾 拒否・中断の判断（diff_guard 等） ここで止めれば、それ以前の情報は揃っている
言葉にすると当たり前に見えますが、動的登録の時代はこの並びを書き下す場所がどこにもありませんでした。書ける場所ができたこと自体が、この変更の実質だったように思います。
from google.antigravity import AgentConfig
HOOK_ORDER = [cost_cap, audit, metrics, diff_guard] # 並びに意味がある
config = AgentConfig( hooks = { "post_tool_use" : HOOK_ORDER }, triggers = { "on_session_end" : [flush_metrics]}, ) 配列に名前を付けて外へ出しておくと、レビューのときに「なぜこの順なのか」を聞かれる場所が1箇所に定まります。宣言を静的に検査するという発想は、agent.md の綴り誤りが権限を広げていた — frontmatter を厳格に検証する lint を書く とも地続きです。
公式に書かれていないハマりどころ
移行のあいだに踏んだもののうち、ドキュメントを読んでも先回りできなかったものを2つ残しておきます。
1つ目は、洗い出しスクリプトが黙って取りこぼす場面です。移行を数ファイル進めた状態で再実行すると、書きかけのファイルが SyntaxError を出します。ここを except SyntaxError: continue で握り潰すと、残件がゼロになったように見えてしまいます。私は一度これで「終わった」と思い込みました。標準エラーへ出す一行を足すだけで回避できますので、必ず入れておくことをお勧めします。
2つ目は、無効化したフックが常駐することの本番運用でのコストです。条件をフックの中へ移した以上、無効なフックも毎回呼ばれます。気になったので測りました。
無効化されたフック1回あたり 0.168 マイクロ秒 (1,000,000回計測)
1セッションで post_tool_use が200回発火し、無効フックが3本ある場合の合計: 100.8 マイクロ秒
1セッションあたり0.1ミリ秒です。私はこの程度なら無視してよいと判断しました。ただし、フックの先頭で設定を読みに行くような実装にすると桁が変わります。早期 return の前に I/O を置かないこと。これは実測する前に決めておいてよい原則だと思います。
移行チェックリスト
scan.py を全ディレクトリに当て、A・B・C の件数を先に把握するC（try の中の登録）があれば、失敗時に止めるか続けるかを先に決める
A を配列へ移す（ここは機械的に進む）
B の条件は宣言側へ持ち込まず、フック内の早期 return へ移す
B のループは生成関数へ切り出し、内包表記で配列へ展開する
並び順を意図して決める （上限の確定 → 記録 → 判断）移行後に落ちるテストが出たら、まず配列の順を疑う
scan.py を CI に入れ、動的登録が再び混入したら落とす
7 と 8 は、実際に踏んでから足したものです。特に 8 は、移行が終わったあとに別ブランチから古い書き方が戻ってきて気づきました。
残っている不便
正直に書いておきますと、宣言へ寄せたことで面倒になった点もあります。
テストのためにフックを1つだけ差し替える、という操作が素直に書けなくなりました。以前は登録を1行足せば済んでいたところです。今は構成を組み立てる小さな関数を挟み、テスト側から配列を差し替える形にしています。
def build_config (hooks = None ): return AgentConfig( hooks = { "post_tool_use" : hooks or HOOK_ORDER }) これで困らなくなりましたが、一手増えたことは事実です。実行時に構成が変わらないという保証と引き換えなので、私はこの取引に納得しております。
順序の非決定性は、動いているうちは誰も気づきません。48.2%という数字を見るまで、私自身、環境のせいだと思い込んでいました。もし手元に「たまに落ちるフック」があるようでしたら、この記事のハーネスをそのまま当てて、自分の構成で何%になるかを一度測ってみていただければと思います。
お読みいただき、ありがとうございました。