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

Managed Agent の非同期ジョブを、投機実行・検証・採用の三段で受け入れる

Gemini API 経由で公開プレビューになった Managed な Antigravity Agent は、サンドボックス内で自律的に計画・実行・検証します。その非同期な成果物を本番に取り込む前に、投機実行・検証・採用の三段で受け止める設計を、実装コードと運用上の落とし穴とともにまとめました。

Managed Agent Gemini API⁴ 非同期処理² Antigravity²⁴⁰

✦ プレミアム記事

Managed な Antigravity Agent が Gemini API で公開プレビューになり、サンドボックス内で計画から実行までを自律でこなすようになりました。

便利さに最初は胸が高鳴ります。けれど少し運用すると、別の不安がやってきます。自律で動くということは、誰の確認も経ずに成果物が出てくるということです。その成果物を、本当にそのまま本番へ流して良いのか。

私自身、複数アプリの運用作業をこの仕組みに任せ始めて、ここで一度立ち止まりました。お伝えしたいのは、自律エージェントの非同期な成果物を、投機実行・検証・採用という三段で受け止める設計です。

自律実行を「いきなり採用」しない

問題の核心は、自律エージェントが「もっともらしいが間違った」成果物を、自信満々で返してくる点にあります。

ファイルを書き換え、ウェブを閲覧し、コードを実行する。その一つひとつは正しく動いていても、最終的な判断が外れていることはあります。これを無検証で本番に採用すると、誤りがそのまま利用者に届きます。

そこで私は、自律実行の出力を必ず「提案」として扱います。提案はまだ採用ではありません。間に検証の段を挟み、合格したものだけを採用へ進める。三段に分けるだけで、自律の暴走が本番に届く確率は大きく下がります。

第一段：投機実行（Propose）

最初の段では、Managed Agent に非同期ジョブを投げ、結果を受け取ります。ここでの出力は本番に触れず、隔離された場所に置きます。

import time
from google import genai
 
client = genai.Client()
 
def propose(task: str, poll_interval: float = 3.0, timeout: float = 300.0):
    """Managed Agent に非同期ジョブを投げ、成果物を提案として受け取る。"""
    job = client.agents.create_run(
        agent="antigravity-preview-05-2026",
        input=task,
    )
    deadline = time.monotonic() + timeout
    while True:
        status = client.agents.get_run(job.id)
        if status.state in ("succeeded", "failed"):
            break
        if time.monotonic() > deadline:
            client.agents.cancel_run(job.id)   # 放置せず必ず止める
            raise TimeoutError(f"job {job.id} がタイムアウトしました")
        time.sleep(poll_interval)
 
    if status.state == "failed":
        raise RuntimeError(f"job {job.id} 失敗: {status.error}")
    return {"job_id": job.id, "artifact": status.output}

ポイントは、タイムアウト時に必ず cancel_run を呼ぶことです。

非同期ジョブを投げっぱなしにすると、サンドボックス側で動き続けて課金だけがかさみます。私はこれで一度、深夜に走り続けたジョブの料金を翌朝に見て青ざめました。投げたら必ず止める経路を用意しておくことを強くお勧めします。

✦

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

この記事の続きを読む

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

この記事で得られること

✦自律エージェントの成果物を無検証で採用しないための三段ゲートの実装

✦検証フェーズで何を機械的に確認し、何を人間に上げるかの線引き

✦非同期ジョブのポーリングで踏みやすい落とし穴と、本番での対処

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

✦

この記事を購入する

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

または

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

第二段：検証（Verify）

提案を受け取ったら、本番に触れる前に検証します。ここが三段設計の心臓部です。

検証は二層に分けます。まず機械的に確認できるものを自動で弾き、機械では判断しきれないものだけを人間に上げます。

def verify(artifact: dict) -> dict:
    """提案を機械検証する。合格・要確認・却下の3値で返す。"""
    issues = []
 
    # 1. 形式チェック：期待する構造を満たすか
    if not artifact.get("files"):
        return {"verdict": "rejected", "reason": "成果物が空です"}
 
    # 2. 安全チェック：触れてはいけない領域に手を出していないか
    for f in artifact["files"]:
        if f["path"].startswith((".env", "secrets/", ".git/")):
            issues.append(f"保護領域への書き込み: {f['path']}")
 
    # 3. 回帰チェック：既存テストが通るか（隔離環境で実行）
    if not run_tests_in_sandbox(artifact):
        issues.append("既存テストが失敗しました")
 
    if any("保護領域" in i for i in issues):
        return {"verdict": "rejected", "reason": "; ".join(issues)}
    if issues:
        return {"verdict": "needs_review", "reason": "; ".join(issues)}
    return {"verdict": "approved", "reason": "全自動チェック通過"}

ここで決定的に大事なのは、検証を「採用とは別の主体」が行うことです。

提案したエージェントに自己採点させると、間違いをもっともらしく正当化します。検証は、生成に関与していない独立したロジックが担う。この分離が、自律実行を安全に使うための前提だと考えています。

第三段：採用（Adopt）

検証で approved になったものだけが、本番へ反映されます。

採用の段では、後から取り消せる形を保ちます。具体的には、採用ごとに記録を残し、いつでも一つ前へ戻せるようにしておきます。

def adopt(artifact: dict, verdict: dict) -> str:
    if verdict["verdict"] != "approved":
        raise PermissionError(f"未承認の成果物は採用できません: {verdict['reason']}")
    rev = commit_to_production(artifact)   # 1コミット=1採用。混ぜない
    record_adoption(job_id=artifact["job_id"], revision=rev)
    return rev