Antigravity CLI の JSON Lines 出力を自作スクリプトで受ける — 部分行と終了コードでつまずいた記録

Antigravity CLI の機械可読出力（JSON Lines）を自前スクリプトに流し込むと、3回に1回ほど JSONDecodeError で止まりました。部分行・終了コード・stderr 混入という3つの罠と、行バッファリングで安定させた実運用の受け口を記録します。

antigravity³⁸³ cli⁶ automation³⁸ json pipeline⁴

✦ プレミアム記事

Antigravity CLI（agy）の出力を別のスクリプトに流し込んで、品質チェックを通してから次の処理へ渡す——という小さなパイプラインを組んだとき、最初の数日は3回に1回ほど json.JSONDecodeError: Expecting value で止まっていました。エージェント自体は正常に走っているのに、その出力を受ける側が壊れる。原因はエージェントでもモデルでもなく、私自身の「出力の受け取り方」にありました。

CLI を対話的に眺めている分には気づきません。出力を機械で受け始めて初めて表に出る問題です。同じところでつまずく方は少なくないと思うので、何が起きていたのか、どう直したのかを順に残しておきます。

機械可読出力は「行」が単位、でも行は途中で届く

Antigravity CLI は、人が読むための整形出力とは別に、機械可読なイベント列を出力できます。バージョンによってフラグ名は異なりますが（agy --help で確認してください）、多くは1行=1イベントの JSON Lines（NDJSON）形式です。agent_started、tool_call、agent_completed のようなイベントが、エージェントの進行に合わせて1行ずつ流れてきます。

ここで最初の誤解がありました。「1行=1 JSON なら、受け取った塊を改行で割って json.loads すればいい」と考えたのです。バッチでまとめて受けるならそれで動きます。問題は、進捗をリアルタイムで見たくてストリーミングで読み始めた瞬間に起きました。

パイプやストリームは「行」という単位を保証してくれません。read(4096) のように固定長で読むと、ちょうど行の途中でぶつ切りになった塊が返ってきます。次の read で残りが届くまで、その行は不完全なままです。私が最初に書いたのは、まさにこの前提を踏み外したコードでした。

import json
import subprocess
 
proc = subprocess.Popen(
    ["agy", "run", "--json", "task.md"],
    stdout=subprocess.PIPE,
    text=True,
)
 
# ❌ 1回の read で「行が途中まで」しか来ないことがある
while True:
    chunk = proc.stdout.read(4096)
    if not chunk:
        break
    for line in chunk.split("\n"):
        event = json.loads(line)  # 部分行・空行で JSONDecodeError
        handle(event)

chunk.split("\n") の最後の要素は、たいてい次のチャンクの先頭と繋がるべき「行の前半分」です。それを単独で json.loads に渡せば当然壊れます。空行（末尾の改行が生む空文字列）でも同じく落ちます。エラーは確率的に出るので、最初は「モデルがたまに変な出力をする」と疑ってしまい、本当の原因にたどり着くまで遠回りしました。

いちばん簡単で正しい受け口は「1行ずつ反復する」

Python のテキストモードのストリームは、for line in stream: で反復すると行の完結を保証してくれます。ぶつ切りの心配をしたくなければ、まずこの形に寄せるのが最短です。あわせて、行バッファリング（bufsize=1）と stderr の分離も最初から入れておきます。

import json
import subprocess
import sys
 
 
def consume(cmd):
    """Antigravity CLI の JSON Lines 出力を1行ずつ安全に受ける。"""
    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,  # 進捗ログを stdout に混ぜない
        text=True,
        bufsize=1,               # 行バッファリング
    )
 
    events = []
    for raw in proc.stdout:      # 1行ずつ・行は必ず完結している
        line = raw.strip()
        if not line:             # 空行はスキップ
            continue
        try:
            events.append(json.loads(line))
        except json.JSONDecodeError:
            # JSON でない行（起動バナー等）が紛れても、握りつぶさず記録だけ残す
            sys.stderr.write(f"skip non-json line: {line[:120]}\n")
 
    code = proc.wait()
    return code, events

for raw in proc.stdout: が肝です。これだけで部分行の問題は消えます。行で読めない事情がない限り、私はまずこの形を推奨します。strip() で前後の空白と改行を落とし、空行を弾き、JSON でない行は捨てつつ標準エラーに記録します。ここで例外を握りつぶして pass だけにすると、後で「なぜか一部のイベントが落ちている」という別の沼にはまります。捨てるにしても、捨てた事実は残す。これは自分への戒めです。

✦

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

この記事の続きを読む

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

この記事で得られること

✦CLI の機械可読出力をパイプで受けたときに行が途中で切れて JSON パースが壊れる問題を、行バッファリングと残余の持ち越しで安定させられるようになります

✦エージェントの成否を「最後の1行」ではなく終了コードと完了イベントの両方で判定する、二重ゲートの設計を自分の自動化に組み込めるようになります

✦stdout（データ）と stderr（進捗ログ）の分離・タイムアウト・実行ごとの一時ディレクトリまで含めた、無人運用に耐える受け口の雛形を手元に持てるようになります

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

✦

この記事を購入する

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

または

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

行で読めない場面では「残余」を自分で持ち越す

ところが、複数のストリームを select で多重化したい場合や、ソケット越しに受ける場合など、行単位の反復に頼れない場面があります。そこでは固定長で読みつつ、改行で割った「最後の半端な行」を次回まで自分でバッファに持ち越す必要があります。ジェネレータにしておくと、呼ぶ側は今まで通り1行ずつ受け取れます。

def iter_json_lines(stream, chunk_size=4096):
    """固定長読みでも、行をまたいだ残余を持ち越して完結した行だけを返す。"""
    buffer = ""
    while True:
        chunk = stream.read(chunk_size)
        if not chunk:
            break
        buffer += chunk
        while "\n" in buffer:
            line, buffer = buffer.split("\n", 1)  # 残りは次回へ持ち越す
            line = line.strip()
            if line:
                yield line
    # ストリーム終端: 改行で終わらない最後の1行も取りこぼさない
    tail = buffer.strip()
    if tail:
        yield tail

ポイントは2つです。1つ目は buffer.split("\n", 1) で先頭の完結した行だけを取り出し、残りを buffer に戻すこと。2つ目は、ストリームが改行で終わらずに閉じたときの最後の tail を必ず拾うことです。この末尾処理を忘れると、最終イベント（多くの場合いちばん大事な完了イベント）だけが静かに消えます。実際に私はこれで一度、「成功しているのに完了と判定されない」という症状を踏みました。

「最後の1行」で成否を決めない — 終了コードと完了イベントの二重判定

受け取りが安定すると、次は成否の判定です。ここでも素朴な実装が罠になります。「最後のイベントが agent_completed なら成功」とだけ書くと、途中でエラーが出ても最後にサマリ行が来ていれば成功と誤判定したり、逆にプロセスがタイムアウトで殺されて完了イベントが出ないまま終了コードだけ非ゼロ、というケースを見落とします。

私が落ち着いたのは、終了コードと完了イベントの両方を見る二重判定でした。どちらか一方では足りません。

def run_and_gate(cmd):
    """終了コードと完了イベントの両方を満たしたときだけ成功とみなす。"""
    code, events = consume(cmd)
 
    completed = any(e.get("type") == "agent_completed" for e in events)
    had_error = any(e.get("type") == "error" for e in events)
 
    # exit 0 でも未完了で抜けることがあるので、両方を確認する
    if code != 0 or had_error or not completed:
        reason = []
        if code != 0:
            reason.append(f"exit={code}")
        if had_error:
            reason.append("error-event")
        if not completed:
            reason.append("no-completion")
        return False, ", ".join(reason), events
 
    return True, "ok", events

なぜここまで慎重にするかというと、無人で回す自動化では「失敗を成功と誤認する」のが最悪だからです。エラーを見逃して次の処理（私の場合は push）まで進んでしまうと、壊れた成果物がそのまま流れていきます。終了コードはプロセスの死に方を、完了イベントはエージェントのやり切り方を表していて、両方が揃って初めて「ちゃんと終わった」と言えます。判定理由を文字列で残しているのは、後でログを見たときに「何で失敗扱いになったのか」を一目で追えるようにするためです。

このあたりの「無人実行で失敗だけを静かに拾う」考え方は、Antigravity CLI を無人で動かすとき、失敗だけを静かに知らせる仕組みでも別角度から整理しています。

stdout と stderr を分ける — 進捗ログが JSON に混ざる事故

二重判定まで組んでもなお JSONDecodeError がたまに出ることがありました。犯人は、CLI が進捗メッセージを標準出力（stdout）に書いていたケースです。データ（JSON Lines）とログ（人間向けの進捗）が同じ stdout に同居すると、受ける側ではどうやっても分離できません。

解決はシンプルで、CLI 側にログを標準エラー（stderr）へ出させ、受ける側でも2つを別々のファイルに分けて受けることです。シェルで組むなら、リダイレクトで明確に分けます。

set -euo pipefail
 
# 実行ごとにユニークな作業ディレクトリを切る（理由は次節）
RUN="$(mktemp -d "${TMPDIR:-/tmp}/agy.XXXXXX")"
trap 'rm -rf "$RUN"' EXIT
 
# stdout(=NDJSON データ) と stderr(=進捗ログ) を別ファイルに分ける
# timeout で無限待ちを防ぐ（モデル応答が固まる事故への保険）
if timeout 1200 agy run --json task.md \
      >"$RUN/out.ndjson" 2>"$RUN/progress.log"; then
  status="ok"
else
  status="failed(exit=$?)"
fi
 
# exit 0 でも未完了で抜けることがあるので、完了イベントの有無も確認する
if ! grep -q '"type":"agent_completed"' "$RUN/out.ndjson"; then
  status="incomplete"
fi
 
echo "status=$status"

>"$RUN/out.ndjson" 2>"$RUN/progress.log" の一行で、データとログがきれいに分かれます。進捗を眺めたいときは progress.log を tail -f すればよく、out.ndjson には JSON 以外の行が混ざりません。timeout を噛ませているのは、ごくまれにモデル応答が固まってプロセスが終わらないことがあり、無人運用ではそれが翌朝までジョブを占有してしまうからです。上限時間を決めておくと、固まったジョブは失敗として切り上げられます。

CI のように端末を持たない環境では、CLI がそもそも機械可読モードに切り替わらず stdout が消える、という別の落とし穴もあります。そちらはAntigravity CLI（agy）を CI で非対話実行するに切り分けの手順をまとめてあります。

実行ごとに作業ディレクトリを切る — 固定名の一時ファイルが招く混入

最後は、受け口そのものより一段外側の話です。出力先を /tmp/agy_out.ndjson のような固定名にしていた時期があり、これがまれに前回ジョブの残骸を次回へ持ち込む事故を起こしました。書き込みが途中で失敗したり、別のジョブと時間が重なったりすると、古い内容が残ったファイルを新しいジョブの出力として読んでしまうのです。

# ❌ 固定名: 書き込み失敗時や実行が重なったとき、前回の残骸が混ざる
# OUT="/tmp/agy_out.ndjson"
 
# ✅ 実行ごとにユニークなディレクトリを切り、終了時に必ず片付ける
RUN="$(mktemp -d "${TMPDIR:-/tmp}/agy.XXXXXX")"
trap 'rm -rf "$RUN"' EXIT
OUT="$RUN/out.ndjson"

mktemp -d で毎回ユニークなディレクトリを作り、trap ... EXIT で正常終了でもエラー終了でも必ず消す。たったこれだけで、ジョブ同士が互いの出力を踏む可能性がなくなります。個人開発で複数のサイトを自動運用していると、別々のジョブがオフピークの時間帯でわずかに重なることがあり、固定名はいつか必ず噛みます。Dolice Labs の自動投稿パイプラインでも、一時ファイルは実行ごとにユニーク名で切り、書き込んだ後に内容を検証する——という二段構えに落ち着きました。

並列でジョブを束ねる設計まで踏み込む場合は、出力の分離と join のやり方をagy の非同期ジョブで夜間タスクを束ねるに整理しています。あわせて読むと、受け口と束ね方が地続きで見えてくるはずです。

まず一本、壊れにくい受け口を用意してから自動化を広げる

エージェントを賢くする話に目が行きがちですが、無人で回す自動化の安定性は、出力をどう受けるかという地味な部分でほとんど決まります。for line in stream: で1行ずつ受け、終了コードと完了イベントの両方で判定し、stdout と stderr を分け、一時ファイルは実行ごとに切る。この4点を入れた受け口を1つ作っておけば、その後どんなタスクを CLI に任せても土台は崩れません。

次の一歩としては、いま動かしている自動化のうち1本だけでいいので、出力の受け取りを for line in proc.stdout: ベースの形に置き換えて、agent_completed の有無を確認するアサーションを足してみてください。私自身、この小さな受け口を整えてから、エージェントに任せられるタスクの幅が静かに広がりました。共に手元の自動化を少しずつ堅牢にしていけたら嬉しいです。