3. エージェントランタイム（PI Embedded）

メッセージ処理ループ、ツール呼び出し、ストリーミング、コンパクション

概要

PI Embedded Agent Runtimeは、OpenClawの中核エンジンです。 @mariozechner/pi-agent-core および @mariozechner/pi-coding-agent SDKをラップし、プロセス内で動作するステートフルなストリーミングエージェントループを実現しています。

「Embedded」の意味：外部APIサービスではなく、OpenClawプロセス内でエージェントが動作します。これにより、永続セッション、統合ツール実行、リアルタイムストリーミング、マルチチャンネル統合が可能になります。

メッセージ処理の全体フロー

メッセージ入力 │ ▼ Phase 1: セットアップ (run.ts:174-305) ├── レーンキューイング（セッション単位 + グローバル） ├── モデル解決（フック: before_model_resolve, before_agent_start） ├── コンテキストウィンドウガード（最小16Kトークン） └── 認証プロファイル解決（ローテーション対応） │ ▼ Phase 2: 実行試行 (attempt.ts:222-1282) ├── ワークスペース・サンドボックスセットアップ ├── ツール生成 → createOpenClawCodingTools() ├── システムプロンプト構築 ├── セッション作成 → createAgentSession() ├── 履歴サニタイズ │ ├── validateAnthropicTurns() / validateGeminiTurns() │ ├── limitHistoryTurns() │ └── sanitizeToolUseResultPairing() ├── サブスクリプション設定 → subscribeEmbeddedPiSession() ├── アクティブラン登録 ├── プロンプト送信 → activeSession.prompt(effectivePrompt) ├── コンパクション待機 └── 結果収集（assistantTexts, toolMetas, usage） │ ▼ Phase 3: リトライ・エラー処理 (run.ts:480-1074) ├── コンテキストオーバーフロー → 自動コンパクション（最大3回） ├── 認証失敗 → プロファイルローテーション ├── 思考レベル非対応 → フォールバック ├── レート制限 / 課金エラー → FailoverError └── タイムアウト → 中断

Phase 1: セットアップの詳細

レーンキューイング

src/agents/pi-embedded-runner/run.ts:177-183

メッセージはセッション固有レーンとグローバルレーンの両方にキューイングされ、セッションごとの逐次処理を保証します。4種類のレーンがあります：

レーン	用途
`Main`	メインセッションの処理
`Cron`	スケジュールタスクの処理
`Subagent`	サブエージェントの処理
`Nested`	ネストされたサブエージェント

認証プロファイル解決

src/agents/pi-embedded-runner/run.ts:307-471

マルチプロファイル認証をサポートし、クールダウン/ローテーション機能があります。あるプロファイルが失敗した場合、自動的に次のプロファイルに切り替わります。

Phase 2: 実行試行の詳細

ツール生成

src/agents/pi-embedded-runner/run/attempt.ts:291-329

createOpenClawCodingTools() がフルツールセットを生成します。ツールは builtInTools と customTools に分割され、マルチレイヤーのポリシーフィルタリングを通過します。

セッション管理

src/agents/pi-embedded-runner/run/attempt.ts:490-589

セッション書き込みロック: acquireSessionWriteLock() で並行書き込みを防止
セッションファイル永続化: SessionManager.open(params.sessionFile) でJSONLファイルから状態を読み込み
履歴サニタイズパイプライン:
- sanitizeSessionHistory() → プロバイダ固有のクリーンアップ
- validateGeminiTurns() / validateAnthropicTurns() → ターン順序の検証
- limitHistoryTurns() → 設定に基づく履歴長の上限
- sanitizeToolUseResultPairing() → 孤立したツール結果の修復

メッセージステアリング

activeSession.steer(text) により、実行中のストリーミングランにメッセージを注入できます。これにより、エージェントの処理中にも追加の指示を送ることが可能です。

ストリーミング / サブスクリプションモデル

src/agents/pi-embedded-subscribe.ts:33-717

サブスクリプションシステムは、SDKイベントとユーザー向け出力の橋渡しを行います。

イベントハンドラディスパッチ

イベント	ハンドラ	実行モード
`message_start`	`handleMessageStart`	同期
`message_update`	`handleMessageUpdate`	同期
`message_end`	`handleMessageEnd`	同期
`tool_execution_start`	`handleToolExecutionStart`	非同期 (best-effort)
`tool_execution_update`	`handleToolExecutionUpdate`	同期
`tool_execution_end`	`handleToolExecutionEnd`	非同期 (best-effort)
`auto_compaction_start`	`handleAutoCompactionStart`	同期
`auto_compaction_end`	`handleAutoCompactionEnd`	同期

テキスト処理パイプライン

生ストリーム │ ▼ stripBlockTags() ← <think>タグ除去、<final>タグ処理 ▼ stripDowngradedToolCallText() ← 履歴アーティファクト除去 ▼ メッセージングツール重複チェック ▼ EmbeddedBlockChunker ← 配信可能なチャンクに分割 ▼ emitBlockChunk() ← onBlockReply()コールバック ▼ リプライディレクティブ解析（メディアURL、音声フラグ、reply-to ID）

状態管理（EmbeddedPiSubscribeState）

assistantTexts[] — 蓄積された最終テキスト
toolMetas[] — ツール呼び出しメタデータ
blockState — <think>/<final>タグのステートフル解析
deltaBuffer / blockBuffer — ストリーミングバッファ
messagingToolSentTexts[] — 重複排除追跡
compactionInFlight / pendingCompactionRetry — コンパクション調整

ツールループ検出

src/agents/tool-loop-detection.ts

エージェントが無限ループに陥ることを防ぐ洗練されたシステムです。直近30回のツール呼び出しをスライディングウィンドウで追跡します。

検出器	閾値	アクション
汎用リピート検出	10回で警告、20回でクリティカル	警告メッセージ注入
ポーリング無進展検出	ポーリングツールの状態変化なし	停止促進
ピンポン検出	交互呼び出しパターン	パターン通知
グローバル回路遮断器	30回同一無進展呼び出し	強制停止

SHA-256ハッシュ（ツール名 + 安定シリアライズされたパラメータ）でパターンマッチングを行います。

コンパクション（コンテキストウィンドウ管理）

src/agents/compaction.ts、src/agents/pi-embedded-runner/compact.ts

2つのモード

モード	トリガー	詳細
自動コンパクション	SDKがコンテキスト上限接近時に発動	auto_compaction_start/endイベントで観測
明示コンパクション	オーバーフロー回復時に呼び出し	compactEmbeddedPiSessionDirect()

コンパクションアルゴリズム

トークン推定: estimateMessagesTokens() + 安全マージン（1.2倍）
適応チャンク比率: メッセージサイズに応じて0.15〜0.40の比率で調整
分割戦略:
- splitMessagesByTokenShare() — 均等トークン分割
- chunkMessagesByMaxTokens() — 最大トークンでチャンク
段階的要約: summarizeInStages()
- パートに分割 → 各パートを要約 → 要約をマージ
- リトライ: 3回試行、500-5000msバックオフ
大型メッセージ除外: コンテキストの50%超のメッセージは要約対象から除外し「[Large message omitted]」として記録
履歴刈り込み: pruneHistoryForContextShare() で最も古いチャンクから削除（最大コンテキストの50%予算）

オーバーフロー回復フロー

コンテキストオーバーフロー検出 │ ▼ 試行が既に自動コンパクション済み？ │ ├── はい → コンパクションなしでリトライ │ └── いいえ ↓ ▼ 明示コンパクション実行 │ ├── 成功 → リトライ │ └── 失敗 ↓ ▼ ツール結果の切り詰め │ ├── 成功 → リトライ │ └── 失敗 ↓ ▼ エラー返却「/resetを使ってください」

エラー処理とリトライ

多層エラー戦略

L1 認証プロファイルローテーション — 失敗プロファイルにクールダウンを設定し次に切り替え
L2 思考レベルフォールバック — サポートされていない思考レベルを段階的にフォールバック
L3 モデルフェイルオーバー — 全プロファイル枯渇時に代替モデルへ切り替え（FailoverError）
L4 コンテキストオーバーフロー回復 — 最大3回のコンパクション試行
L5 タイムアウト処理 — AbortControllerによるキャンセル

「Embedded」ならではの特徴

単純なAPI呼び出しと異なる点：

プロセス内実行 — 同一Node.jsプロセスで動作し、状態を共有
永続セッション — JSONLファイルに完全なセッション状態を永続化
並行ラン管理 — グローバル ACTIVE_EMBEDDED_RUNS マップでアクティブランを追跡
レーンベース並行制御 — セッションレーン + グローバルレーンで競合を防止
セッション書き込みロック — ファイルレベルのロック
統合ツール実行 — ワークスペース・サンドボックス・ファイルシステムへのフルアクセス
ストリーミング統合 — WebSocket経由のリアルタイムイベント発行
プラグイン/フックシステム — 全ステージにライフサイクルフック
マルチチャンネルサポート — 同一ランタイムで全チャンネルを処理
サンドボックス分離 — ツール呼び出しのオプショナルサンドボックス実行

← システムプロンプトメモリシステム →