Grok、Gemini、ChatGPT／Codex、Claude Code を 1 本の自動化パイプラインにつなぐとき、決め手はつなぐモデルの数ではありません。状態を進めるのは誰か、ゲートを守るのは誰か、信頼できないコンテンツはどこで止まるのか、です。実際に動作を確かめた watchdog オーケストレーター、認証付きの FastAPI Broker、引き継ぎプロトコルのテンプレートを示し、よくある実装の落とし穴——状態が IN_PROGRESS のまま止まる、レビュアーに書き込み権限を渡す、`git diff HEAD~1` でコミットを見落とす、TOTP シークレットを Redis に置く——を一つずつ正します。エージェント間のプロンプトインジェクション、`ANTHROPIC_API_KEY` の認証優先順位の罠、そして「2 つ目のアカウント」では 2 つ目の視点が得られない理由も取り上げます。
対象読者：ソフトウェアエンジニア、システムアーキテクト、AI DevTools 開発者、技術系ギーク
主なテーマ：異種 AI エージェントの協調パターン、アカウントをまたぐ状態の分離、コンテキスト引き継ぎプロトコル、ローカル中継サービス（Broker）と自動化パイプラインの構築
本稿の Claude Code のフラグは、すべて Claude Code 2.1.278 の
claude --helpと公式ドキュメントに突き合わせて確認しています。第 4 節のオーケストレーター、Broker、ポーリングスクリプトは実際に動かしました。モックのclaude実行ファイルを使って 5 つのシナリオ（一発で通過、テストが失敗してから通過、レビューで差し戻されてから承認、進捗なし、3 回連続の差し戻し）を流し、状態遷移はすべて想定どおりでした。各社の CLI は更新が速いので、読む時点では手元の--helpの出力を基準にしてください。
1. 背景と設計原則
1.1 単一エージェントの限界
ソフトウェア開発が複雑になるほど、単一の LLM や単一の AI エージェント端末に頼るやり方は、次のようなボトルネックに突き当たります。
- コンテキストの汚染と忘却（Context Pollution）：会話が長くなると、エージェントは序盤に与えられたアーキテクチャ上の制約を少しずつ失い、意味のドリフトや幻覚を起こします。プロジェクトを通じて守るべき制約は、起動のたびに読み直されるプロジェクト指示ファイル（
CLAUDE.md、AGENTS.md）に書くべきで、チャット履歴に期待するものではありません。 - クォータと課金の境界（Rate Limits & Quota）：長時間にわたって集中的にコードを生成すると、単一プラットフォームの利用上限にすぐ達します。性質の違う作業を別々のプラットフォームに振り分けること自体が負荷分散になります。ただしそれは「同じサービスのアカウントを何個も作って順番に回す」こととは別の話で、これは 1.3 節で改めて取り上げます。
- 単一モデルの思考の死角：推論、リファクタリング、大量のログ解析、リアルタイムの情報収集で、モデルごとに得手不得手があります。1 つのモデルですべてをうまくこなすのは困難です。
- 実行と監査の同一化（Self-Audit Risk）：同じエージェントにコードを書かせ、セキュリティ監査までさせると、自分の死角を見逃しがちです。
1.2 アカウント／プラットフォームをまたぐ疎結合アーキテクチャ
アカウントとプラットフォームをまたぐエージェントパイプラインの核となる考え方は、「暗黙の会話記憶」を「明示的な状態資産」に変えることです。
1[1] 調査 ── Grok：リアルタイムの Web / X 検索；Gemini：長文ドキュメントとマルチモーダル資料の読み込み
2 │ 成果物：docs/research/*.md（すべての結論に出典 URL を付ける）
3 ▼
4[2] 仕様 ── ChatGPT / Codex：推論、インターフェース定義、テスト項目
5 │ 成果物：インターフェースファイル、テスト項目、.pipeline/HANDOVER.md
6 │ ★ 人による確認ポイント：アーキテクチャの決定はここで確定させる
7 ▼
8[3] 実装 ── Claude Code（profile acc_a）：リポジトリの読み書き、テスト実行
9 │ ゲート：オーケストレーター自身が npm test を実行し、通ったときだけコミット
10 ▼
11[4] レビュー ── Claude Code（profile acc_b、読み取り専用）または別ベンダーのモデル
12 APPROVE → 完了；CHANGES_REQUESTED → [3] に差し戻し
13
14状態バス：Git リポジトリ（コードとコミット履歴）＋ .pipeline/（HANDOVER.md、pipeline_state.json。コミットしない）**状態バス（State Bus）と引き継ぎプロトコル（Handover Protocol）**を導入すれば、異なるアカウント・異なるプラットフォームのエージェントが、ログイン認証情報や API キーを一切共有せずに、非同期でタスクをリレーできます。実際の「バス」は 2 本あります。コードとコミット履歴は Git を、パイプラインの状態は .pipeline/ ディレクトリを通ります（.gitignore に追加するのを忘れずに）。
1.3 「アカウントをまたぐ」ことで実際に何が解決するのか
レビュアーを「2 つ目のアカウント」で動かし、独立した視点を得ようとするやり方がよく見られます。ここでは 2 つの事柄を切り分ける必要があります。
- 独立したコンテキストに 2 つ目のアカウントは要りません。
claude -pの呼び出しは毎回まっさらなセッションで、実装者の会話履歴は見えません。Claude Code に組み込まれたサブエージェント（.claude/agents/*.md、tools:で使えるツールを絞れます）も、それぞれ独立したコンテキストウィンドウを持ちます。 - 2 つ目のアカウントは 2 つ目の視点をもたらしません。 同じモデルをアカウントだけ変えて使っても、学習で身についた傾向や死角は変わりません。本当の意味でのクロス監査をしたいなら、レビュー段階には別ベンダーのモデルを使うべきです（第 5 節 Step 4 の末尾にある代替案を参照）。
では、設定ディレクトリ（CLAUDE_CONFIG_DIR）やアカウントを分けることに意味はないのかというと、あります。ただし理由は容量の拡大ではなく分離です。業務の組織と個人のアカウントを分ける、パイプライン専用に別課金で支出上限を設定できる API キーを使う、ロールごとの認証情報が漏れたときの影響範囲を互いに閉じ込める、監査ログをロールごとに区別できるようにする、といったことです。
もう 1 つ、はっきり書いておくべき線引きがあります。Anthropic は Claude Code の法務・コンプライアンスのページで、Pro／Max プランの公表されている利用枠は「通常の個人利用」を前提としていること、製品やサービスを構築する開発者は Claude Console の API キー認証を使うべきことを明記しています。複数のサブスクリプションアカウントをクォータのプールにして、24 時間 365 日無人で回るパイプラインで順番に使うやり方は、この前提に反するうえ、いつ止まってもおかしくない基盤の上にパイプラインを載せることになります。無人の自動化を動かすなら API キーを使い、Console で支出上限を設定してください。他のベンダーにもそれぞれ規約があり、本番投入の前に一通り読んでおく価値があります。
2. 異種エージェントのマトリクスと役割分担
各モデルの強みを最大限に活かすには、それぞれの本来の能力に沿ってパイプライン上の役割をはっきり決める必要があります。
| プラットフォーム | 主な強み | 役割と典型的な成果物 | 自動化の手段 |
|---|---|---|---|
| Grok | リアルタイムの Web と X の検索、技術動向の追跡 | 調査担当：技術選定レポート、最新の API 変更メモ（出典付き） | Grok Build CLI（grok -p、ベータ）；または xAI API（OpenAI 互換）＋サーバー側の web_search／x_search ツール |
| Gemini | 超長コンテキスト、PDF・画像・動画などのマルチモーダル解析 | 分析担当：レガシーリポジトリの要約、大量のログや長文ドキュメントのダイジェスト | Gemini CLI（gemini -p） |
| ChatGPT／Codex | 論理的推論、複雑なアルゴリズム設計、仕様策定 | アーキテクト：インターフェース仕様、テスト項目、HANDOVER.md | Codex CLI（codex exec）または OpenAI API |
| Claude Code（acc_a） | ターミナル操作、複数ファイルの変更、テスト駆動の反復 | メイン実装担当：アプリケーションコード、ユニットテスト | claude -p |
| Claude Code（acc_b）または別ベンダー | 独立したコンテキスト、読み取り専用の権限 | レビュアー：レビューレポートと、承認／差し戻しの明確な判定 | 読み取り専用設定の claude -p；または codex exec --sandbox read-only |
補足が 2 点あります。
- 本稿ではモデルのバージョンを固定しません。 o1／o3、GPT-4o、grok-3 といった、この種のチュートリアルでよく見かける型番は、2026 年の時点ではどれも各社の主力ではありません。grok-3 に至っては 2026 年 5 月に提供終了となり、リクエストは新しいモデルに自動で転送されます。モデル名はアーキテクチャ図に書き込むものではなく、パイプライン設定のパラメータとして扱うべきです。
- 「長いコンテキスト」はもはや Gemini だけの売りではありません。 主要なフラッグシップモデルは、数十万から 100 万トークン規模のコンテキストウィンドウを備えるのが当たり前になっています。長文ドキュメントを Gemini に任せる実際的な理由は、マルチモーダル解析の能力、独立したクォータ、そして「別のモデルに読ませる」こと自体がもたらす視点の違いです。
3. コンテキスト引き継ぎプロトコル（Context Handover Standard）
エージェント間でスムーズに引き継ぐには、標準化された「引き継ぎファイルのプロトコル」が必要です。しかも、人が読めることと機械が解析できることを両立させなければなりません。
まず、2 種類のコンテキストを分けて考えます。
- 静的な制約（コーディング規約、ディレクトリの境界、禁止事項）は、プロジェクトのルートにある
AGENTS.mdに書きます。Codex、Cursor、Jules などはこれをネイティブに読みます。Gemini CLI はデフォルトではGEMINI.mdを読みますが、.gemini/settings.jsonで"context": {"fileName": "AGENTS.md"}を設定できます。Claude Code はCLAUDE.mdに@AGENTS.mdと 1 行書けば取り込めます。1 つのファイルを、すべてのエージェントで共有するわけです。 - 動的な引き継ぎ（今回のタスクがどこまで進み、次に何をするか）は
.pipeline/HANDOVER.mdに書き、各段階の終わりに更新します。
3.1 構造化された HANDOVER.md
各エージェントは、自分の担当段階を終えるときに .pipeline/HANDOVER.md を出力（または更新）します。
1# Agent Handover Protocol v1.1
2
3## 1. タスクの基本情報 (Task Metadata)
4
5- **Source Agent**: codex（仕様段階）
6- **Target Agent**: claude-code:acc_a（実装段階）
7- **Timestamp**: 2026-09-21T11:30:00+09:00
8- **Pipeline ID**: pipe_feat_auth_v2_88f9a
9- **Base Commit**: 3f2a9c1e7b4d
10
11## 2. 目的と範囲 (Goal & Scope)
12
13- **Goal**: 既存の JWT ログインフローに、TOTP による二要素認証（2FA）を追加する。
14- **In-Scope**: `src/auth/`, `src/services/totpService.ts`, `src/middleware/auth.ts`, `tests/auth/`
15- **Out-of-Scope**: UI コンポーネント。`src/config/jwt.ts` の署名アルゴリズムと鍵ローテーション。
16
17## 3. 完了した作業 (Completed Actions)
18
19- [x] `IAuthService` インターフェースを定義した（`docs/specs/auth_spec.md` を参照）。
20- [x] TOTP ライブラリとして `otplib` v13 を選定した。
21
22## 4. コンテキストと制約 (Context & Constraints)
23
24- **重要なファイル**:
25 - `docs/specs/auth_spec.md`: 厳守すべきインターフェース定義。
26 - `src/config/jwt.ts`: 参照のみ。変更しないこと。
27- **既知の落とし穴 (Pitfalls)**:
28 - TOTP シークレットは長期間有効な認証情報であり、暗号化したうえでデータベースに永続化する。
29 Redis やプロセスメモリだけに置いてはならない。Redis の役割は 2 つだけ：ユーザーごとに
30 最後に検証を通過したタイムステップの記録（RFC 6238 §5.2 は、同じタイムステップ内の OTP を
31 2 回目に受け付けてはならないと定めている）と、失敗回数によるレート制限。
32 - otplib v13 は全面的に書き直された版で、`authenticator` のエクスポートは削除され、`verify()` は
33 非同期になりオブジェクトを返す（`result.valid` を読む）。ネット上の v12 の例の多くはそのままでは
34 動かない。リプレイ対策には `afterTimeStep` オプションが使える。
35 - `auth.ts` のミドルウェアは独自の `AppError` クラスに依存しており、素の `Error` を投げないこと。
36
37## 5. 受け入れ基準 (Acceptance Criteria)
38
39- `npm test` がすべて通り、`totpService` の行カバレッジが 90% 以上。
40- 同じタイムステップ内で同じ OTP を 2 回目に送信すると 401 を返す。
41
42## 6. 受け手エージェントの次のアクション (Next Actions for Receiver)
43
441. `npm install otplib@^13` を実行する。
452. `docs/specs/auth_spec.md` に従って `src/services/totpService.ts` を実装する。
463. `totpService` のユニットテストを書く。
474. git commit はしないこと。コミットはテスト通過後にオーケストレーターが行う。
48
49## 7. 未解決の問題 (Open Questions)
50
51- リカバリーコードの生成と保存の方式は未定。この段階では実装しない。このテンプレートには、掘り下げておきたい設計がいくつかあります。どれも、引き継ぎ文書でよくある間違いに対応しています。
- 範囲と指示は一致させる。 In-Scope に
src/services/がないのに、次のアクションでそこに新しいファイルを作るよう指示すると、エージェントは範囲を踏み越えるか、行き詰まるかのどちらかになります。 - 引き継ぎ文書に誤った設計判断を運ばせない。 「マルチインスタンス構成では TOTP シークレットを Redis に保存しなければならない」という書き方をよく見かけます。しかし TOTP シークレットはパスワードと同じく長期間有効な認証情報で、暗号化してデータベースに置くべきものです。Redis はキャッシュなので、キーが追い出されることもあれば、永続化が有効になっていないこともあります。Redis に置くのに向いているのは、リプレイ防止の記録とレート制限のカウンターです。引き継ぎ文書の誤った制約は、下流のエージェントに「決定済みの事項」として忠実に実行されてしまいます。
- 「JWT ベースの二要素認証」は混乱した表現です。 JWT はセッションを、TOTP は第 2 の要素を担い、両者は直交しています。目的は「JWT ログインフローに TOTP を追加する」と書くべきです。
- 受け入れ基準と未解決の問題は省略しない。 受け入れ基準がなければ、「完了」はエージェントの自己申告でしかありません。未解決の問題のリストがなければ、エージェントがあなたの代わりに勝手に決めてしまいます。
- タイムスタンプは ISO 8601 で書き、3.3 節の diff に使う
Base Commitを記録します。
3.2 機械向けの状態機械 pipeline_state.json
Markdown に加えて、アトミックな状態変化を記録する JSON ファイルを用意し、オーケストレーションのスクリプト（Watcher／Broker）が次の手を判断できるようにします。
1{
2 "pipeline_id": "pipe_feat_auth_v2_88f9a",
3 "current_stage": "IMPLEMENTATION",
4 "status": "AWAITING_EXECUTION",
5 "base_commit": "3f2a9c1e7b4d",
6 "retry_count": 0,
7 "updated_at": "2026-09-21T11:30:00+09:00",
8 "history": [
9 {
10 "stage": "RESEARCH",
11 "agent": "grok",
12 "status": "COMPLETED",
13 "output_artifacts": ["docs/research/2fa.md"]
14 },
15 {
16 "stage": "SPECIFICATION",
17 "agent": "codex",
18 "status": "COMPLETED",
19 "output_artifacts": ["docs/specs/auth_spec.md", ".pipeline/HANDOVER.md"]
20 }
21 ]
22}status は AWAITING_EXECUTION → IN_PROGRESS →（次の段階の AWAITING_EXECUTION、COMPLETED、HUMAN_INTERVENTION_REQUIRED のいずれか）と遷移します。next_agent_target のようなフィールドはあえて置いていません。どの段階をどのエージェントが実行するかはオーケストレーターの設定であり、状態ファイルにも書くと、食い違いうる情報源が 2 つできるだけです。
さらに重要なのは、このファイルに書き込めるのは誰かです。上流（人、または第 4 節の Broker のポーリングスクリプト）は、状態を AWAITING_EXECUTION にすることしかしません。それ以降の遷移はすべてオーケストレーターが行い、エージェント自身は状態ファイルに書き込みません。LLM は不正な JSON を書き出すこともあれば、段階を飛ばすことも、テストが落ちているのに「完了しました」と宣言することもあります。状態機械は、エージェントの自己申告ではなく、終了コードとテスト結果に基づいて進めるべきです。
3.3 Git によるコードの意味を伝える引き継ぎ
コードを変更する段階では、変更そのものが最も強いコンテキストになります。仕様段階を例に取ると、最後に次のようにコミットします。
1git add docs/specs/auth_spec.md src/auth/IAuthService.ts
2git commit -m "feat(auth): define 2FA spec and IAuthService interface
3
4- Add IAuthService interface
5- Store TOTP secrets encrypted in the DB; Redis only for replay guard and rate limits
6
7Pipeline-ID: pipe_feat_auth_v2_88f9a
8Handover-To: claude-code:acc_a"細かい点をいくつか挙げます。
- コミットするパスは明示的に列挙します。
git add .だと、.envやビルド成果物、.pipeline/までまとめて取り込んでしまいます。 Pipeline-IDは Git の trailer として書きます（コミットメッセージの最後の段落）。あとからgit log --grep="Pipeline-ID: pipe_feat_auth_v2_88f9a"で、1 本のパイプラインのコミットをすべて見つけられます。- 下流では
git diff HEAD~1ではなくgit diff <base_commit>...HEADを使います。 実装段階で複数のコミットができることは珍しくなく、「差し戻し→再実装」を経ればなおさらで、HEAD~1では最後の 1 つしか見えません。3 点ドットの記法は、マージベースからHEADまでの変更をすべて比較します。
diff は下流に何を変えたかを伝え、コミットメッセージと HANDOVER.md はなぜ変えたかを伝えます。どちらも欠かせません。ただし diff には変更されなかった呼び出し側が含まれないので、レビュアーは関連ファイルを読む必要があります。第 4 節のレビュー設定で Read と Grep を残しているのはそのためです。
4. アカウント／プラットフォームをまたぐ通信バス（Inter-Agent Communication Bus）
互いに独立したエージェント同士を連鎖的に起動するには、自動化された通信バスが必要です。まず前提を 1 つはっきりさせておきます。Web 版の ChatGPT、Gemini、Grok は、スクリプトから起動することも、結果をリポジトリに書き込むこともできません。 自動化パイプラインに組み込むには、各社の CLI か API に切り替えます。OpenAI の Codex CLI（codex exec）、Google の Gemini CLI（gemini -p）、xAI の Grok Build CLI（grok -p、現在ベータ）または xAI API です。完全には自動化できない段階を「人が成果物をリポジトリに置く」半自動のステップとして残すのは、まったく合理的な判断です。
方式 A：共有ファイルシステムと軽量な Watchdog リスナー
1 台のマシンで複数の端末を使う場面に最も向いた方式です。Python スクリプトが .pipeline/pipeline_state.json の変化を監視し、次のエージェントを自動で起動します。
オーケストレーションスクリプト agent_orchestrator.py
1#!/usr/bin/env python3
2"""
3agent_orchestrator.py — .pipeline/pipeline_state.json を監視し、段階ごとにローカルの
4Claude Code を起動する。
5
63 つの設計原則：
71. 状態を遷移させるのはオーケストレーターだけ。エージェントは成果物を出して終了し、状態ファイルには触れない。
82. ゲートは終了コードとオーケストレーター自身が実行するテストで判定し、エージェントの自己申告は信用しない。
93. 状態ファイルは必ずアトミックな置き換えで書き込み、読み手が中途半端な JSON を読むことはない。
10オーケストレーターは同時に 1 インスタンスだけ動かす前提。
11"""
12import json
13import os
14import subprocess
15import tempfile
16import threading
17import time
18from datetime import datetime
19from pathlib import Path
20
21from watchdog.events import FileSystemEventHandler
22from watchdog.observers import Observer
23
24PIPELINE_DIR = Path(".pipeline").resolve()
25STATE_FILE = PIPELINE_DIR / "pipeline_state.json"
26# "~" はコード側で絶対パスに展開する：クォート内や Python 文字列の "~" はシェルが展開しない
27PROFILES = Path.home() / ".claude_profiles"
28MAX_RETRIES = 3
29AGENT_TIMEOUT_SEC = 30 * 60
30
31STAGES = {
32 "IMPLEMENTATION": {
33 "profile": "acc_a",
34 "prompt": (
35 ".pipeline/HANDOVER.md を読み、そこで定められた範囲内で実装を完成させ、ユニットテストも揃えてください。"
36 ".pipeline/feedback.md があれば、まずそこに挙げられた問題を一つずつ解決してください。"
37 ".pipeline/ 配下のファイルは変更せず、git commit も実行しないでください。"
38 ),
39 "flags": [
40 "--max-turns", "60",
41 "--max-budget-usd", "5",
42 "--permission-mode", "acceptEdits",
43 "--allowedTools", "Bash(npm test *)", "Bash(npx tsc *)", "Bash(git diff *)", "Bash(git status)",
44 ],
45 },
46 "REVIEW": {
47 "profile": "acc_b",
48 "prompt": (
49 "あなたは独立したセキュリティレビュアーです。標準入力は今回の変更の完全な diff です。"
50 "コンテキストを補うためにリポジトリのファイルを読むのは構いませんが、何も変更してはいけません。"
51 "重点的に確認すること：接続リーク、未処理の Promise rejection、EVAL で文字列連結して組み立てた Lua スクリプト、"
52 "複数スロットにまたがるマルチキー操作、そして package.json の scripts や CI 設定など実行内容を変える変更。"
53 "レビュー結果は Markdown で出力し、"
54 "最後の行は必ず VERDICT: APPROVE または VERDICT: CHANGES_REQUESTED にしてください。"
55 ),
56 "flags": [
57 "--max-turns", "30",
58 "--max-budget-usd", "2",
59 # dontAsk：確認が必要な操作はすべて自動で拒否し、下のホワイトリストの読み取り専用ツールだけを許可する
60 "--permission-mode", "dontAsk",
61 "--allowedTools", "Read", "Grep", "Glob",
62 "--disallowedTools", "Edit", "Write", "NotebookEdit", "Bash", "WebFetch", "WebSearch",
63 ],
64 },
65}
66
67wake = threading.Event()
68
69
70class StateFileHandler(FileSystemEventHandler):
71 # 購読するのは「書き込み」系のイベントだけ。Linux の watchdog は読み取りだけでも opened/closed を発行するので、
72 # on_any_event を使うとオーケストレーターが状態を読むたびに自分自身を起こしてしまう。
73 # 「一時ファイルに書いて rename」で保存するツールが多く、その場合対象ファイルには moved しか届かず modified は来ない。
74 def on_modified(self, event):
75 self._check(event.src_path)
76
77 def on_created(self, event):
78 self._check(event.src_path)
79
80 def on_moved(self, event):
81 self._check(event.dest_path)
82
83 def _check(self, path):
84 if Path(path).resolve() == STATE_FILE:
85 wake.set()
86
87
88def load_state():
89 for _ in range(10):
90 try:
91 return json.loads(STATE_FILE.read_text(encoding="utf-8"))
92 except FileNotFoundError:
93 return None
94 except json.JSONDecodeError:
95 time.sleep(0.2) # 外部の書き手がまだ書き終えていない可能性がある
96 raise RuntimeError(f"{STATE_FILE} を解析できない状態が続いている")
97
98
99def save_state(state):
100 state["updated_at"] = datetime.now().astimezone().isoformat(timespec="seconds")
101 fd, tmp = tempfile.mkstemp(dir=PIPELINE_DIR, suffix=".tmp")
102 with os.fdopen(fd, "w", encoding="utf-8") as f:
103 json.dump(state, f, ensure_ascii=False, indent=2)
104 os.replace(tmp, STATE_FILE) # 同一ファイルシステム内でのアトミックな置き換え
105
106
107def git(*args):
108 return subprocess.run(
109 ["git", *args], capture_output=True, text=True, check=True
110 ).stdout
111
112
113def run_claude(stage, stdin_text=None):
114 spec = STAGES[stage]
115 # -p モードでは環境に ANTHROPIC_API_KEY があれば必ずそれが優先され、設定ディレクトリによる分離が効かなくなる。
116 # パイプラインを API 課金で動かしたい場合は、このフィルタを外して専用のキーを明示的に渡す
117 env = {k: v for k, v in os.environ.items() if k not in ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN")}
118 env["CLAUDE_CONFIG_DIR"] = str(PROFILES / spec["profile"])
119 # プロンプトは -p の直後に置き、--allowedTools のような可変長の引数は末尾にまとめる
120 cmd = ["claude", "-p", spec["prompt"], "--output-format", "json", *spec["flags"]]
121 # 入力がないときも空の stdin を渡す：さもないと子プロセスがオーケストレーターの stdin を継承し、
122 # TTY でない環境では入力を待ち続けることがある
123 proc = subprocess.run(
124 cmd, env=env, input=stdin_text or "", capture_output=True, text=True,
125 timeout=AGENT_TIMEOUT_SEC,
126 )
127 try:
128 result = json.loads(proc.stdout)
129 except json.JSONDecodeError:
130 result = {"is_error": True, "result": proc.stderr[-2000:]}
131 result["exit_code"] = proc.returncode
132 return result
133
134
135def record(state, stage, status, result):
136 state.setdefault("history", []).append({
137 "stage": stage,
138 "agent": f"claude-code:{STAGES[stage]['profile']}",
139 "status": status,
140 "session_id": result.get("session_id"),
141 "num_turns": result.get("num_turns"),
142 "cost_usd": result.get("total_cost_usd"),
143 "at": datetime.now().astimezone().isoformat(timespec="seconds"),
144 })
145
146
147def advance(state, stage, next_stage, result):
148 record(state, stage, "COMPLETED", result)
149 state["current_stage"] = next_stage
150 state["status"] = "COMPLETED" if next_stage == "DONE" else "AWAITING_EXECUTION"
151 save_state(state)
152
153
154def fail(state, stage, feedback, result, back_to=None, give_up=False):
155 # retry_count はパイプライン全体の失敗回数：テスト失敗、レビューでの差し戻し、タイムアウトをすべて数える
156 state["retry_count"] = state.get("retry_count", 0) + 1
157 record(state, stage, "FAILED", result)
158 (PIPELINE_DIR / "feedback.md").write_text(feedback, encoding="utf-8")
159 if give_up or state["retry_count"] >= MAX_RETRIES:
160 state["status"] = "HUMAN_INTERVENTION_REQUIRED"
161 notify(f"Pipeline {state['pipeline_id']} は {stage} 段階で停止：{feedback[:200]}")
162 else:
163 state["current_stage"] = back_to or stage
164 state["status"] = "AWAITING_EXECUTION"
165 save_state(state)
166
167
168def notify(message):
169 print(f"[!] {message}", flush=True) # 本番では webhook による通知に差し替える（6.3 節を参照）
170
171
172def implement(state):
173 result = run_claude("IMPLEMENTATION")
174 if result.get("is_error") or result["exit_code"] != 0:
175 return fail(state, "IMPLEMENTATION", str(result.get("result", "")), result)
176 tests = subprocess.run(["npm", "test"], capture_output=True, text=True)
177 if tests.returncode != 0:
178 return fail(state, "IMPLEMENTATION", tests.stdout[-4000:] + tests.stderr[-2000:], result)
179
180 git("add", "-A", "--", ".", ":(exclude).pipeline")
181 tree = git("write-tree").strip()
182 # HEAD や前回とまったく同じツリー：エージェントは堂々巡りしており、続けてもトークンを燃やすだけ
183 if tree in (state.get("last_tree"), git("rev-parse", "HEAD^{tree}").strip()):
184 return fail(state, "IMPLEMENTATION", "今回のラウンドでは新しい変更がなく、進捗なしと判定した。", result, give_up=True)
185 state["last_tree"] = tree
186 git("commit", "-m", f"feat: implement {state['pipeline_id']}\n\nPipeline-ID: {state['pipeline_id']}")
187 (PIPELINE_DIR / "feedback.md").unlink(missing_ok=True)
188 advance(state, "IMPLEMENTATION", "REVIEW", result)
189
190
191def review(state):
192 diff = git("diff", f"{state['base_commit']}...HEAD", "--", ".", ":(exclude).pipeline")
193 result = run_claude("REVIEW", stdin_text=diff)
194 report = str(result.get("result", ""))
195 (PIPELINE_DIR / "audit_report.md").write_text(report, encoding="utf-8")
196 if result.get("is_error") or result["exit_code"] != 0:
197 return fail(state, "REVIEW", report, result)
198 if report.rstrip().endswith("VERDICT: APPROVE"):
199 return advance(state, "REVIEW", "DONE", result)
200 # 判定行がない場合も差し戻しとして扱う：レビューしきれていない変更を通すより、1 ラウンド多く回すほうがよい
201 fail(state, "REVIEW", report, result, back_to="IMPLEMENTATION")
202
203
204HANDLERS = {"IMPLEMENTATION": implement, "REVIEW": review}
205
206
207def tick():
208 state = load_state()
209 while (
210 state
211 and state.get("status") == "AWAITING_EXECUTION"
212 and state.get("current_stage") in HANDLERS
213 ):
214 stage = state["current_stage"]
215 state["status"] = "IN_PROGRESS" # この後にプロセスが落ちたら、手動で AWAITING_EXECUTION に戻す
216 save_state(state)
217 print(f"[>] {state['pipeline_id']}: {stage}", flush=True)
218 try:
219 HANDLERS[stage](state)
220 except Exception as exc: # タイムアウトや git の失敗なども、リトライ回数に数える
221 fail(state, stage, f"orchestrator error: {exc!r}", {})
222 state = load_state()
223
224
225if __name__ == "__main__":
226 PIPELINE_DIR.mkdir(exist_ok=True)
227 observer = Observer()
228 observer.schedule(StateFileHandler(), str(PIPELINE_DIR), recursive=False)
229 observer.start()
230 print(f"[+] Orchestrator を起動、{STATE_FILE} を監視中", flush=True)
231 wake.set() # 起動時にまず 1 回確認し、停止中にたまったタスクを拾う
232 try:
233 while True:
234 wake.wait(timeout=30) # ファイルイベントでの起床が主、30 秒ごとのポーリングは保険
235 wake.clear()
236 tick()
237 except KeyboardInterrupt:
238 pass
239 finally:
240 observer.stop()
241 observer.join()ネット上でよく見かける最小限の例は、数十行程度です。on_modified のコールバックで JSON を読んで状態を書き換え、そのまま --dangerously-skip-permissions 付きのエージェントを subprocess.run で起動する、というものです。考え方は正しいのですが、そのまま真似すると次のような落とし穴にはまります。上の実装はそれを一つずつ処理しています。
- ファイルイベントは見た目より複雑です。 watchdog 6.0.0（Linux）で実際に試したところ、ファイルを読むだけで
FileOpenedEventとFileClosedNoWriteEventが発生しました。「一時ファイルに書いて rename」で保存すると、対象ファイルにはFileMovedEventが 1 つ届くだけで、on_modifiedはまったく呼ばれません。普通の書き込み 1 回でon_modifiedが複数回呼ばれることもあります。そのため、ここでは書き込み系のイベントだけを購読し、threading.Eventで連続するイベントを 1 回の処理にまとめています。 - ディスパッチでイベントスレッドをブロックしない。 30 分かかるかもしれないエージェントを watchdog のコールバック内で実行すると、イベントの配送がまるごと止まります。ここではコールバックは「起こす」だけで、実際の処理はメインループで行います。
- 状態は入れっぱなしにしない。 素朴な実装は状態を
IN_PROGRESSにしたあと、終了コードも確認せず、状態を先に進めることもせず、JSON の書き換えを完全にエージェント任せにします。エージェントがその手順を 1 回飛ばしただけで、パイプラインはエラーも出さずに永遠に止まります。 - レビュアーに書き込み権限はそもそも要りません。 diff は stdin で渡し、レポートは stdout で受け取り、
audit_report.mdにはオーケストレーターが書き込みます。レビュアーには読み取り専用を求めながら--dangerously-skip-permissionsで起動する、というのはこの種の例でよく見られる矛盾です。 --output-format jsonはsession_id、num_turns、total_cost_usdなどのフィールドを返します。オーケストレーターはこれをhistoryに記録するので、各ラウンドにいくらかかり、何ターン使ったかを後から追跡できます。
2 つの設定ディレクトリでは、それぞれ最初に一度だけ対話的にログインしておく必要があります。CLAUDE_CONFIG_DIR="$HOME/.claude_profiles/acc_a" claude を実行し、/login を実行してください。公式ドキュメントによれば、CLAUDE_CONFIG_DIR を設定すると、認証情報ファイル（macOS ではキーチェーンの項目）がディレクトリごとに分かれます。完全に無人で動かすなら、claude setup-token で有効期間 1 年の OAuth トークンを生成して CLAUDE_CODE_OAUTH_TOKEN で渡すか、ANTHROPIC_API_KEY を使って API 課金にします。優先順位には注意が必要です。-p モードでは、環境に ANTHROPIC_API_KEY があれば必ずそれが使われ、OAuth トークンやサブスクリプションのログインより優先されます。上のコードが子プロセスの環境からこれを取り除いているのはそのためです。
方式 B：軽量な Python HTTP Broker（マシン間の中継役）
エージェントが別々の物理マシンや CI 環境に分かれている場合は、FastAPI の Broker を置けます。
1# broker_server.py — マシン間の引き継ぎを中継する最小限のサービス（単一プロセス、デモ用）
2import os
3import secrets
4from collections import defaultdict, deque
5
6from fastapi import Depends, FastAPI, Header, HTTPException
7from pydantic import BaseModel
8
9app = FastAPI(title="Multi-Agent Handover Broker")
10TOKEN = os.environ["BROKER_TOKEN"] # トークンが設定されていなければ起動しない
11
12
13def require_token(authorization: str = Header(default="")):
14 if not secrets.compare_digest(authorization.encode(), f"Bearer {TOKEN}".encode()):
15 raise HTTPException(status_code=401)
16
17
18class HandoverPayload(BaseModel):
19 pipeline_id: str
20 sender_agent: str
21 receiver_agent: str
22 stage: str
23 handover_markdown: str
24 git_commit: str # 共有リモートにプッシュ済みのコミット：マシン間では参照だけを渡し、ローカルパスは渡さない
25
26
27# 受け手ごとに FIFO キューを 1 本。プロセスのメモリ上にしかない：再起動で消え、複数ワーカーでも動かせない
28queues: dict[str, deque] = defaultdict(deque)
29
30
31@app.post("/api/v1/handover", dependencies=[Depends(require_token)])
32async def enqueue(payload: HandoverPayload):
33 queues[payload.receiver_agent].append(payload)
34 return {"status": "ACK", "queued": len(queues[payload.receiver_agent])}
35
36
37@app.get("/api/v1/poll/{agent_id}", dependencies=[Depends(require_token)])
38async def poll(agent_id: str):
39 queue = queues.get(agent_id)
40 if not queue:
41 return {"has_task": False, "data": None}
42 return {"has_task": True, "data": queue.popleft()}
43
44
45if __name__ == "__main__":
46 import uvicorn
47
48 # 待ち受けはローカルホストのみ。マシン間でアクセスするなら VPN / Tailscale や TLS 付きのリバースプロキシの内側に置く
49 uvicorn.run(app, host="127.0.0.1", port=8080)Claude Code を動かすマシンでは、ポーリングスクリプトを 1 つ用意し、受け取ったタスクを方式 A の状態ファイルに書き出して、あとはオーケストレーターに任せます。
1# poll_worker.py — Claude Code のあるマシンで動かし、Broker のタスクを方式 A のオーケストレーターに渡す
2import json
3import os
4import subprocess
5import time
6from pathlib import Path
7
8import httpx
9
10BROKER = os.environ.get("BROKER_URL", "http://127.0.0.1:8080")
11HEADERS = {"Authorization": f"Bearer {os.environ['BROKER_TOKEN']}"}
12AGENT_ID = "claude-code-acc-a"
13STATE_FILE = Path(".pipeline/pipeline_state.json") # .pipeline/ は .gitignore に入れておく
14
15
16def busy():
17 if not STATE_FILE.exists():
18 return False
19 return json.loads(STATE_FILE.read_text("utf-8"))["status"] in ("AWAITING_EXECUTION", "IN_PROGRESS")
20
21
22while True:
23 time.sleep(10)
24 if busy(): # オーケストレーターがまだ作業中なので、新しいタスクは取りに行かない
25 continue
26 try:
27 resp = httpx.get(f"{BROKER}/api/v1/poll/{AGENT_ID}", headers=HEADERS, timeout=10)
28 resp.raise_for_status()
29 except httpx.HTTPError as exc:
30 print(f"[worker] broker に接続できない：{exc}", flush=True)
31 continue
32 task = resp.json()
33 if not task["has_task"]:
34 continue
35
36 data = task["data"]
37 subprocess.run(["git", "fetch", "origin"], check=True)
38 subprocess.run(["git", "checkout", "-B", f"pipeline/{data['pipeline_id']}", data["git_commit"]], check=True)
39 STATE_FILE.parent.mkdir(exist_ok=True)
40 (STATE_FILE.parent / "HANDOVER.md").write_text(data["handover_markdown"], encoding="utf-8")
41 state = {
42 "pipeline_id": data["pipeline_id"],
43 "current_stage": data["stage"],
44 "status": "AWAITING_EXECUTION",
45 "base_commit": data["git_commit"],
46 "retry_count": 0,
47 "history": [],
48 }
49 tmp = STATE_FILE.with_suffix(".tmp")
50 tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
51 tmp.replace(STATE_FILE) # アトミックな置き換え。方式 A のオーケストレーターがすぐに起動するこの Broker は短いものですが、4 つの点は意図的にそうしています。
- 認証と待ち受けアドレス。 多くの例は
0.0.0.0で待ち受け、認証もまったくありません。しかし Broker が配送したタスクは、最終的にコマンドを実行できるエージェントに渡ります。マシンのシェルをサブネット全体に公開しているのと同じことです。 - キューは本当にキューでなければならない。
message_queue[receiver] = payloadで実装した「キュー」では、2 つ目のタスクが 1 つ目を黙って上書きします。ここでは受け手ごとに FIFO キューを 1 本用意しています。 - マシン間でローカルパスを渡さない。 ペイロードに
docs/spec.mdのようなローカルパスを入れても、別のマシンでは意味を持ちません。ここではgit_commitを渡し、成果物そのものは Git のリモートを通じて同期します。 - 配送のセマンティクスを把握しておく。
pollは取り出した時点で削除するので「最大 1 回（at-most-once）」です。ワーカーがタスクを受け取った直後に落ちれば、そのタスクは失われます。本番では確認応答の仕組みがあるキューに置き換えてください。たとえば Redis Streams のコンシューマーグループ（XREADGROUP+XACK）や、SQS の可視性タイムアウトです。
呼び出し方：いわゆる「ChatGPT／Gemini ノード」の実体は、自分で書くスクリプトです。Codex CLI や Gemini CLI、各社の API を呼び出して作業を済ませ、結果を Git のリモートにプッシュしてから、/api/v1/handover に POST を送ります。Web 版の ChatGPT が社内ネットワークの Broker にリクエストを送ることはできません。
方式 C：CLI の標準入出力（Stdio／Pipe）のリダイレクト
シェルのレベルでは、Unix のパイプを使って、あるプラットフォームの出力をそのまま別のエージェントに渡せます。
1#!/usr/bin/env bash
2# pipeline_pipe.sh
3set -euo pipefail
4mkdir -p .pipeline
5
6echo "=== ステップ 1：Gemini CLI で長文ドキュメントを分析 ==="
7gemini -p "@docs/legacy_architecture.pdf に書かれた主要な性能ボトルネックを詳しく要約し、簡潔な Markdown で出力してください" \
8 > .pipeline/gemini_summary.md
9
10echo "=== ステップ 2：要約を stdin で Claude Code に渡してリファクタリング ==="
11claude -p "標準入力はアーキテクチャ上のボトルネックの要約です。これに基づいて src/legacy_module.js をリファクタリングし、終わったら npm test を実行してください。" \
12 --permission-mode acceptEdits \
13 --allowedTools "Bash(npm test *)" \
14 --max-turns 40 \
15 < .pipeline/gemini_summary.md間違えやすい点が 2 つあります。Google 公式 CLI の実行ファイル名は gemini で、リポジトリ名の gemini-cli ではありません。ファイルはプロンプト内の @パス で取り込みます。set -euo pipefail は、ステップ 1 が失敗した時点でパイプライン全体を止め、空の要約が下流に渡るのを防ぎます。
ステップ 2 ではあえて --dangerously-skip-permissions を使わず、「編集は自動承認、許可するのはテストコマンドだけ」にしています。理由は権限の大きさだけではありません。パイプは、上流のモデルの出力をそのまま下流のエージェントへの指示に変えてしまうのです。 Gemini が読んでいるのは 1 つの PDF であり、その中のどんな一節も、このパイプを通ってあなたのマシン上のコマンドになりえます。これがまさに 6.2 節で扱う問題です。
5. 実践：プラットフォームをまたぐエンドツーエンドのエージェント開発パイプライン
以下は、実際のシナリオに沿った半自動のワークフローです。最初の 2 段階は人が起動して確認し、残りの 2 段階は第 4 節のオーケストレーターに任せます。
シナリオ：エンタープライズ向け Redis キャッシュ層のリファクタリングと独立したセキュリティレビュー
Step 1：調査段階（Grok）
- Prompt：
"Redis Cluster 構成における Node.js のクライアント選定と再接続戦略を調査してください。すべての結論に出典リンクを付け、各クライアントライブラリの現在のメンテナンス状況も明記してください。" - 成果物：
docs/research/redis_client.md
プロンプトを「ioredis の最適な再接続戦略を調べて」とはあえて書いていません。それでは答えを前提にしてしまいます。先に選定を問うと、調査からは重要な事実が返ってきます。ioredis の README には、メンテナンスは「best-effort」（できる範囲で）であり、新規プロジェクトには node-redis を推奨すると書かれているのです。調査を飛ばしてコーディングエージェントに記憶だけで書かせれば、学習データに最も多く出てくるライブラリを選ぶ可能性が高いでしょう。
「すべての結論に出典を付ける」という要求も形式的なものではありません。検索型のモデルも API の変更をでっち上げることがあり、リンクのない結論は次の段階に持ち込むべきではありません。
Step 2：仕様とテスト設計の段階（ChatGPT／Codex）
- 入力：
docs/research/redis_client.md - Prompt：
"チーフアーキテクトとして、調査レポートに基づいて CacheManager の TypeScript インターフェースを定義してください。インターフェースに特定のクライアントライブラリの型を露出させてはいけません。ユニットテストの項目リストと .pipeline/HANDOVER.md も出力してください。" - 成果物：
src/cache/ICacheManager.tstests/specs/cache_spec.md.pipeline/HANDOVER.md
「インターフェースにクライアントライブラリの型を出さない」という一文によって、node-redis か ioredis かという議論は、先送りも撤回もできる実装の詳細になります。ここはパイプライン全体で最も人による確認を入れる価値がある場所でもあります。アーキテクチャの判断を誤れば、後続のすべての段階がその誤りを非常に真面目に実行してしまうからです。確認が済んだらコミットし、そのコミットの SHA を base_commit として状態ファイルに書き込み、状態を AWAITING_EXECUTION にします。あとはオーケストレーターが引き継ぎます。
Step 3：コア実装の段階（Claude Code、profile acc_a）
オーケストレーターが実行するコマンドは、次と同等です。
1CLAUDE_CONFIG_DIR="$HOME/.claude_profiles/acc_a" claude -p \
2 ".pipeline/HANDOVER.md と src/cache/ICacheManager.ts を読み、CacheManager を実装してユニットテストも揃えてください。git commit は実行しないでください。" \
3 --output-format json \
4 --max-turns 60 --max-budget-usd 5 \
5 --permission-mode acceptEdits \
6 --allowedTools "Bash(npm test *)" "Bash(npx tsc *)"ここで CLAUDE_CONFIG_DIR="~/.claude_acc_a" ではなく $HOME と書いている点に注意してください。ダブルクォート内の ~ はシェルが展開しないので、プログラムにはリテラルの ~/.claude_acc_a が渡ります。それで動くかどうかは、この変数を読むすべてのプログラムが自前で ~ を処理するかどうか次第です。そこに賭けるより、最初から $HOME と書くほうが確実です。また、権限のフラグは省略できません。-p モードでは「許可」を押す人がいないので、あらかじめ許可しておかなければ、ファイルの編集もテストの実行もすべて拒否され、エージェントは実質的に何も変更できません。
処理の流れ：
- エージェントが
HANDOVER.mdの範囲と制約を読み込みます。 src/cache/CacheManager.tsとそのユニットテストを生成し、自分でnpm testを実行して通るまで繰り返します。- エージェントが終了したあと、オーケストレーターが自ら
npm testをもう一度実行します。数に入るのはこちらの結果です。 - テストが通れば、オーケストレーターが変更をコミットし（
Pipeline-IDの trailer 付き）、状態をREVIEWに進めます。
なぜエージェント自身に git add . && git commit させたり、pipeline_state.json を書き換えさせたりしないのでしょうか。プロンプトに一文書き漏らすだけで、パイプラインはエラーも出さずにそこで止まってしまうからです。さらに git add . は、コミットすべきでないファイルまでまとめて取り込んでしまいます。
Step 4：クロスセキュリティレビューの段階（Claude Code、profile acc_b）
1git diff "$BASE_COMMIT"...HEAD | CLAUDE_CONFIG_DIR="$HOME/.claude_profiles/acc_b" claude -p \
2 "あなたは独立したセキュリティレビュアーです。標準入力は今回の変更の完全な diff です。接続リーク、監視されていない error イベントと未処理の Promise rejection、EVAL で文字列連結して組み立てた Lua スクリプト、複数スロットにまたがるマルチキー操作を重点的に確認してください。最後の行には VERDICT: APPROVE または VERDICT: CHANGES_REQUESTED と出力してください。" \
3 --permission-mode dontAsk \
4 --allowedTools "Read" "Grep" "Glob" \
5 --disallowedTools "Edit" "Write" "Bash" \
6 --max-turns 30 > .pipeline/audit_report.md- 成果物：
.pipeline/audit_report.md。最後の行がAPPROVEならパイプラインは完了です。CHANGES_REQUESTEDの場合、あるいは判定行そのものがない場合は、レポートがfeedback.mdに書き込まれ、タスクは Step 3 に差し戻されます。
レビューの重点に「Redis インジェクション」を入れていないのは意図的です。RESP プロトコルはコマンドを引数の配列として送るので、SQL インジェクションのような意味での連結の脆弱性は存在しません。Redis で本当に危ないのは、文字列連結で組み立てた Lua スクリプト（EVAL）と、名前空間のプレフィックスがないキーです。Cluster 構成ではさらにマルチキー操作の確認が必要です。関係するキーが同じスロットにないとコマンドは CROSSSLOT エラーで失敗するので、{user:42} のようなハッシュタグで同じスロットに寄せる必要があります。
本当の意味でモデルの多様性が欲しいなら、このステップを別ベンダーのモデルに置き換えます。たとえば Codex CLI に読み取り専用のサンドボックスでレビューさせる方法です：codex exec --sandbox read-only "git diff $BASE_COMMIT...HEAD の変更をレビューしてください……"。同じモデルをアカウントだけ変えて使っても、得られるのはまっさらなコンテキストであって、2 つ目の視点ではありません。
6. エンジニアリング上の安全性、サンドボックス化、耐障害性
マルチエージェントのパイプラインは、複雑になるほど耐障害性とセキュリティ制御が重要になります。
6.1 権限とアカウントのサンドボックス化
-
設定ディレクトリの分離（Profile Sandboxing） 環境変数でロールごとの認証状態を分け、ロールをまたいで認証情報を共有しないようにします。
alias claude-dev='CLAUDE_CONFIG_DIR="$HOME/.claude_profiles/acc_a" claude' alias claude-audit='CLAUDE_CONFIG_DIR="$HOME/.claude_profiles/acc_b" claude'alias をシングルクォートで定義すると、変数は使うたびに展開されます。
-
ロールごとの最小権限
ロール 権限モード 許可 禁止 実装担当 acceptEditsファイル編集； Bash(npm test *)、Bash(npx tsc *)それ以外の Bash コマンド（ -pモードでは承認する人がいないので拒否される）レビュアー dontAskRead、Grep、GlobEdit、Write、Bash、ネットワーク系ツール -
権限ルールはガードレールであって、セキュリティ境界ではない
Bash(npm test *)が許可するのはコマンドのプレフィックスですが、npm testが実際に何を実行するかはpackage.jsonのscripts.testで決まり、実装担当はまさにそのファイルを変更できます。さらに困ったことに、オーケストレーター自身のテストゲートもそれを実行します。したがって、- 実装担当とテストゲートは、どちらもコンテナ、Dev Container、VM の中で動かすべきです。そこには本番の認証情報を置かず、ネットワークの出口は必要なものだけを許可リストで通します。Claude Code に組み込まれたサンドボックス（設定項目
sandbox.enabled。macOS、Linux、WSL2 で Bash にファイルシステムとネットワークの分離を提供します）も、防御の 1 層として使えます。 claude --helpは--dangerously-skip-permissionsについて「Recommended only for sandboxes with no internet access」（インターネットにアクセスできないサンドボックスでのみ推奨）と説明しています。ホストマシン上ですべてのエージェントにこれを付けるのは、マルチエージェントのチュートリアルで最もよく見かけ、かつ最も危険なやり方です。- レビュアーの確認項目には、
package.jsonの scripts と CI 設定の変更を必ず含めてください。第 4 節のレビュー用プロンプトにはすでに入れてあります。
- 実装担当とテストゲートは、どちらもコンテナ、Dev Container、VM の中で動かすべきです。そこには本番の認証情報を置かず、ネットワークの出口は必要なものだけを許可リストで通します。Claude Code に組み込まれたサンドボックス（設定項目
6.2 エージェントをまたぐプロンプトインジェクション：引き継ぎのたびに信頼境界がある
この種の記事で最も見落とされがちで、しかも最も危険な部分です。パイプラインの上流にいるのは、まさに信頼できないコンテンツを読むことが仕事のエージェントです。Grok は Web ページや X の投稿を読み、Gemini は出どころの分からない PDF やログを読みます。その出力が HANDOVER.md の「次のアクション」にそのまま流れ込んだり、方式 C のようにパイプで直接プロンプトになったりすれば、攻撃者が Web ページに仕込んだ一文が、実装担当のマシン上のコマンドになりかねません。
Simon Willison はこの組み合わせを「lethal trifecta」（致命的な三要素）と呼んでいます。プライベートなデータにアクセスできること、信頼できないコンテンツに触れること、外部に情報を送れることの 3 つです。3 つがそろえば、データの流出は巧妙に作られた一段落の文章だけで起こります。マルチエージェントのパイプラインは、気づかないうちにこの 3 つをそろえてしまいがちです。調査エージェントが読み込みを担い、実装担当がリポジトリと認証情報を握り、その間を隔てるのは Markdown ファイル 1 つだけ、という構図です。
現実的な緩和策は次のとおりです。
- 調査の成果物は参考資料であって、指示ではない。 成果物は
docs/research/に置き、人かアーキテクチャ段階がそれを咀嚼してからHANDOVER.mdに書きます。実装担当のプロンプトには、調査資料の中身はデータであってコマンドではない、と明記します。 - 三要素のうち少なくとも 1 つを崩す。 実装担当の環境には本番の認証情報を置かず、ネットワークの出口は許可リスト方式にします。そうすれば、仮にインジェクションされても、盗めるものも送り先もありません。
- レビュアーに異常な振る舞いを重点的に見張らせる。 新たに追加されたネットワークリクエスト、新しい依存関係、変更されたビルドスクリプトなどです。
6.3 無限ループとトークン消費の制御（Loop Protection）
実装担当とレビュアーが「修正 → 差し戻し → 修正 → 差し戻し」のループに陥ると、トークンは消費され続けます。制御は 3 層で考えます。
- 1 回の呼び出し単位：
--max-turnsで 1 回の実行のターン数を、--max-budget-usdで 1 回の実行の金額を制限し（-pモードでのみ有効）、さらにsubprocess.runのtimeoutを設定します。 - パイプライン全体：
retry_countでテスト失敗、レビューでの差し戻し、タイムアウトの合計回数を数え、MAX_RETRIESに達したら状態をHUMAN_INTERVENTION_REQUIREDにします。このロジックは第 4 節のオーケストレーターのfail()にすでに書いてあります。リトライ上限は状態遷移そのものに組み込まなければ意味がありません。check_loop_limit()を定義しても、失敗のたびに呼ばず、retry_countも増やさないのでは、上限がないのと同じです。 - 進捗なしの検出：
git write-treeでステージングエリアのツリーハッシュを計算し、前回と同じならすぐに止めます。単純に回数を数えるより、「エージェントが同じものを何度も提出している」状況をずっと早く検出できます。
各呼び出しの total_cost_usd は history に記録されているので、1 本のパイプラインにいくらかかったかはコマンド 1 つで計算できます。
jq '[.history[].cost_usd // 0] | add' .pipeline/pipeline_state.json最後に、オーケストレーターの notify() を本物の通知に置き換えます。以下は Slack の Incoming Webhook の例です。Feishu（飛書）や DingTalk（釘釘）のカスタムボットでは、JSON の構造が違うだけです（Feishu は {"msg_type": "text", "content": {"text": ...}}、DingTalk は {"msgtype": "text", "text": {"content": ...}}）。
1import json
2import os
3import urllib.request
4
5
6def notify(message):
7 print(f"[!] {message}", flush=True)
8 url = os.environ.get("ALERT_WEBHOOK_URL")
9 if not url:
10 return
11 req = urllib.request.Request(
12 url,
13 data=json.dumps({"text": message}).encode(),
14 headers={"Content-Type": "application/json"},
15 )
16 try:
17 urllib.request.urlopen(req, timeout=10)
18 except OSError as exc: # 通知が送れなくても、オーケストレーターを道連れにしてはいけない
19 print(f"[!] 通知の送信に失敗：{exc}", flush=True)7. まとめと、これからの Agent Mesh
アカウントとプラットフォームをまたぐ AI エージェントの連携ワークフローは、AI 支援開発が**「単独で戦う AI アシスタントの時代」から「複数エージェントが協調するパイプラインの時代」**へ移りつつあることを示しています。
Grok（リアルタイムの情報収集）、Gemini（長文ドキュメントとマルチモーダルの理解）、ChatGPT／Codex（推論と仕様策定）、**Claude Code（ターミナルでの実行とリファクタリング）**を組み合わせ、標準化された引き継ぎプロトコル（HANDOVER.md）と軽量なディスパッチバスを添えれば、スループットが高く、分離が効いていて、段階同士が互いにレビューし合う自動化されたソフトウェア生産ラインを組み立てられます。ただし本稿で繰り返し確かめてきたのは、このラインの信頼性がつなぐモデルの数では決まらないということです。決め手になるのは、状態を進めるのは誰か、ゲートを守るのは誰か、信頼できないコンテンツはどこで止まるのかです。
これからについて言えば、相互接続の標準はすでに十分具体的になっており、Unix ドメインソケット上に独自プロトコルを発明する必要はありません。
- **MCP（Model Context Protocol）**は、ツールとコンテキストをエージェントにつなぎます。Claude Code 自身も
claude mcp serveで MCP サーバーとして動作し、他のエージェントからツールとして呼び出せます。 - A2A（Agent2Agent）プロトコルは、エージェント同士の通信を担います。2026 年 8 月に Linux Foundation 傘下の Agentic AI Foundation（AAIF）に加わり、仕様のバージョンは 1.0 です。エージェントは
/.well-known/agent-card.jsonで自分の能力を名刺（Agent Card）として公開します。 - AGENTS.md も AAIF が管理しており、ツールをまたいでプロジェクトの制約を共有する事実上の標準になりつつあります。
振り返ってみると、本稿で手書きした pipeline_state.json の状態機械と Broker は、本質的には A2A のタスク（Task）のライフサイクルと成果物（Artifact）の受け渡しを手作業で実装したものです。HUMAN_INTERVENTION_REQUIRED は、A2A の input-required 状態にそのまま対応します。まず簡略版を自分の手で組み、それから標準プロトコルに移行すれば、各フィールドがなぜ存在するのかがはっきり分かるはずです。
導入チェックリスト：
.pipeline/を.gitignoreに追加し、設定ディレクトリごとに一度/loginで対話的にログインしておく。- 状態を遷移させるのはオーケストレーターだけ。ゲートは終了コードとテスト結果で判定する。
- レビュアーは読み取り専用。diff は stdin で渡し、レポートは stdout で受け取る。
- 実装担当はサンドボックスで動かし、本番の認証情報に触れさせず、ネットワークの出口は許可リスト方式にする。
- すべての呼び出しに
--max-turnsと--max-budget-usdを、すべてのパイプラインにリトライ上限を設定する。 - 仕様段階のあとに、人による確認を 1 回入れる。
- 無人のパイプラインでは、複数のサブスクリプションアカウントを回すのではなく、API キーを使う。
コメント
コメント (0)