[確実なハンドオフ可視性のためのsessions_sendイベントフック機能リクエスト] - Feature Request: sessions_send Event Hook for Guaranteed Handoff Visibility

現在の制限の現れ方

エージェントがsessions_sendを使用してセッション間連携を行う際、一般的に以下のような障害パターンが観察されます：

確認なしのサイレント処理

# エージェントAが連携を開始
sessions_send(
    target_session="project-alpha",
    payload={"task": "review_pr_442", "priority": "high"}
)
# エージェントBがサイレントに処理 — 対象トピックにメッセージが投稿されない
# ユーザーに連携が発生した気配がない

脆弱なプロンプトレベルの強制

現在の回避策は、すぐに失敗するプロンプト命令に依存しています：

# エージェントAのプロンプト内（コンテキスト圧縮や高負荷時に無視されやすい）：
# "sessions_sendを呼び出した後は、必ず対象トピックに可視メッセージを投稿してください。"

# 現実：エージェントはコンテキストを圧縮し、この指示を失う可能性がある、または：
# 1. sessions_sendを呼び出す ✓
# 2. 可視メッセージを投稿するのを忘れる ✗
# 3. ユーザーは連携状況を確認できない

送信者への配信確認がない

# エージェントAが連携を送信
result = sessions_send(...)

# resultでは以下の確認ができない：
# - メッセージがキューに入れられたか
# - 対象セッションが存在するか
# - 配信が試行されたか
# エージェントAは不確実な状態で動作する

ユーザー側の症状

• ユーザーはセッション間連携が成功したかどうか判断できない
• エージェント間での作業分散の監査証跡がない
• マルチエージェントワークフローが不透明で信頼できないに見える
• 連携失敗のデバッグにはログの手動検査が必要

アーキテクチャのギャップ分析

現在のsessions_sendの実装には根本的な設計上の制約があります：

1. ファイア・アンド・フォgettweetのメッセージ配信

# 現在の実装（概念図）
def sessions_send(target_session, payload, ...):
    queue_message(target_session, payload)
    return {"status": "queued"}  # フックなし、副次的効果なし

この関数はペイロードを配信しますが、以下の拡張ポイントがありません：

• 配信時の副次的効果
• セッション横断の通知
• 監査ログ記録

2. トランスポートとプレゼンテーションの分離

sessions_sendトランスポート層はTelegramプレゼンテーション層から切り離されています：

┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ Agent A │ │ Gateway │ │ Agent B │ │ sessions_send │─────▶│ (transport) │─────▶│ (processing) │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ │ │ ▼ ▼ ▼ ┌─────────────────┐ (no receipt) (no hook point) │ Telegram Bot │ │ (presentation) │ └─────────────────┘

3. プロンプトレベルの強制は本質的に信頼性が低い

現在の回避策は、保証できないLLMの動作に依存しています：

• コンテキストウィンドウの圧力が命令の喪失を引き起こす
• 圧縮により重要な動作プロンプトが剥離される可能性がある
• エージェントの自律性により、プロンプトは示唆であり制約ではない
• sessions_sendとメッセージ投稿の間に原子性がない

4. セッション横断の調整のためのイベントシステムがない

イベントフックの欠如により：

• Gatewayは配信時に副次的効果をトリガーできない
• 送信者への配信確認の機会がない
• 自動肯定応答には手動のエージェント実装が必要
• 宣言的な設定パスがない

提案実装：sessionReceiveイベントフック

ステップ1：Gateway設定でフックを構成する

OpenClaw Gateway設定に以下を追加します：

{
  "hooks": {
    "sessionReceive": {
      "autoAcknowledge": {
        "enabled": true,
        "message": "📨 {sender}からの連携を受け取りました — 処理中です。",
        "channel": "last",
        "topic_id": "{incoming_topic}"
      },
      "deliveryConfirmation": {
        "enabled": true,
        "confirm_to_session": "{sender_session}",
        "confirm_message": "✅ {target_session}に{timestamp}に配信しました"
      }
    }
  }
}

ステップ2：設定パラメータの理解

• `enabled`: フックを有効にするブール値
• `message`: プレースホルダー付きのテンプレート文字列：
  • `{sender}` — 送信エージェント/セッションの名前
  • `{sender_session}` — 送信者のセッションID
  • `{target_session}` — このセッションのID
  • `{incoming_topic}` — メッセージが到着したTelegramトピックID
  • `{timestamp}` — ISO 8601形式の配信タイムスタンプ
• `channel`: 肯定応答の送信先（「last」、特定のトピックID、またはnull）
• `confirm_to_session`: 配信 receipt 送信先のセッションID
• `confirm_message`: Receipt テンプレートメッセージ

ステップ3：回避策の実装（現在のPythonスクリプト）

ネイティブフックサポートが実装されるまでは、提供されたhandoff.pyスクリプトをデプロイします：

# handoff.py — 連携可視性を保証するスクリプト

import os
import requests
from datetime import datetime

class HandoffManager:
    def __init__(self, bot_token: str, gateway_url: str):
        self.bot_token = bot_token
        self.gateway_url = gateway_url
        self.telegram_api = f"https://api.telegram.org/bot{bot_token}"
    
    def send_with_guaranteed_visibility(
        self,
        target_session: str,
        target_topic_id: int,
        payload: dict,
        sender_name: str = "System"
    ) -> dict:
        """
        保証された可視的な肯定応答付きで連携を実行します。
        """
        # ステップ1: 対象トピックに先に可視メッセージを投稿
        confirmation_msg = f"📨 {sender_name}からの連携を受信"
        self._post_telegram_message(target_topic_id, confirmation_msg)
        
        # ステップ2: 対象セッションにペイロードを配信
        delivery_result = self._deliver_to_session(target_session, payload)
        
        # ステップ3: 配信肯定応答を送信者に返す（セッションが提供された場合）
        if sender_session := payload.get("_sender_session"):
            self._send_confirmation(sender_session, target_session)
        
        return {
            "status": "completed",
            "visible_posted": True,
            "payload_delivered": True,
            "confirmation_sent": bool(payload.get("_sender_session"))
        }
    
    def _post_telegram_message(self, topic_id: int, text: str) -> dict:
        """特定のTelegramトピックにメッセージを投稿します。"""
        return requests.post(
            f"{self.telegram_api}/sendMessage",
            json={
                "chat_id": os.environ["TARGET_CHAT_ID"],
                "message_thread_id": topic_id,
                "text": text
            }
        ).json()
    
    def _deliver_to_session(self, session_id: str, payload: dict) -> dict:
        """Gateway sessions_send経由でペイロードを配信します。"""
        return requests.post(
            f"{self.gateway_url}/sessions/{session_id}/receive",
            json=payload,
            headers={"Authorization": f"Bearer {os.environ['GATEWAY_TOKEN']}"}
        ).json()
    
    def _send_confirmation(self, sender_session: str, target_session: str):
        """送信者セッションに配信receiptを送信します。"""
        confirmation = {
            "type": "handoff_confirmation",
            "target": target_session,
            "timestamp": datetime.utcnow().isoformat(),
            "status": "delivered"
        }
        requests.post(
            f"{self.gateway_url}/sessions/{sender_session}/receive",
            json=confirmation,
            headers={"Authorization": f"Bearer {os.environ['GATEWAY_TOKEN']}"}
        )

ステップ4：OpenClawエージェントとの統合

# エージェントのツール実装またはミドルウェア内

from handoff import HandoffManager

# マネージャー 초기化（環境変数または設定から）
hm = HandoffManager(
    bot_token=os.environ["TELEGRAM_BOT_TOKEN"],
    gateway_url=os.environ["GATEWAY_URL"]
)

def safe_sessions_send(target_session: str, target_topic: int, payload: dict):
    """
    保証された可視性を持つsessions_sendのドロップイン置換。
    """
    enriched_payload = {
        **payload,
        "_sender_session": current_session_id,  # 確認を有効化
        "_handoff_type": "cross_agent"
    }
    
    result = hm.send_with_guaranteed_visibility(
        target_session=target_session,
        target_topic_id=target_topic,
        payload=enriched_payload,
        sender_name=current_agent_name
    )
    
    if not result["status"] == "completed":
        raise HandoffError(f"配信失敗: {result}")
    
    return result

回避策実装の検証ステップ

ステップ1：単一連携の可視性をテスト

# 連携テストを実行
python -c "
from handoff import HandoffManager
import os

hm = HandoffManager(
    bot_token=os.environ['TELEGRAM_BOT_TOKEN'],
    gateway_url=os.environ['GATEWAY_URL']
)

result = hm.send_with_guaranteed_visibility(
    target_session='test-agent-01',
    target_topic_id=42,
    payload={'task': 'verify_handoff', 'test': True},
    sender_name='TestHarness'
)

print(f'Status: {result[\"status\"]}')
print(f'Visible Posted: {result[\"visible_posted\"]}')
print(f'Payload Delivered: {result[\"payload_delivered\"]}')
"

期待される出力：

Status: completed
Visible Posted: True
Payload Delivered: True
Payload Confirmation: True

ステップ2：Telegramメッセージが表示されることを確認

対象Telegramトピックで可視確認メッセージを確認します：

# 対象トピックに表示される期待されるメッセージ：
📨 Handoff incoming from TestHarness

ステップ3：送信者トピックで配信確認を検証

送信者のセッション/トピックで配信receiptを確認します：

# 送信者トピックに表示される期待されるメッセージ：
✅ Delivered to test-agent-01 at 2024-01-15T10:30:00+00:00

ステップ4：Gateway配信ログを検証

# 配信確認のGatewayログを確認
curl -s -H "Authorization: Bearer $GATEWAY_TOKEN" \
  "$GATEWAY_URL/sessions/test-agent-01/history?limit=5" | jq '.[] | select(.type=="handoff_confirmation")'

期待される結果：

{
  "type": "handoff_confirmation",
  "target": "test-agent-01",
  "timestamp": "2024-01-15T10:30:00+00:00",
  "status": "delivered"
}

ステップ5：エンドツーエンドのマルチエージェントフローテスト

# 完全なマルチエージェントワークフローをシミュレート
python -c "
from handoff import HandoffManager
import os

hm = HandoffManager(
    bot_token=os.environ['TELEGRAM_BOT_TOKEN'],
    gateway_url=os.environ['GATEWAY_URL']
)

# Agent A → Agent B → Agent C チェーン
sessions = ['agent-a', 'agent-b', 'agent-c']
topics = [10, 20, 30]

for i in range(len(sessions) - 1):
    result = hm.send_with_guaranteed_visibility(
        target_session=sessions[i + 1],
        target_topic_id=topics[i + 1],
        payload={
            'task': f'handoff_{i}',
            '_sender_session': sessions[i],
            '_chain_position': i + 1
        },
        sender_name=f'Agent-{chr(65+i)}'
    )
    assert result['status'] == 'completed', f'Handoff {i} failed'

print('Chain verification: ALL PASSED')
"

環境と設定の落とし穴

1. 必須環境変数の欠落

# 必要だが不足しがちな設定：
# TELEGRAM_BOT_TOKEN      - Bot APIトークン
# GATEWAY_URL             - Gateway基本URL
# GATEWAY_TOKEN           - Gateway認証
# TARGET_CHAT_ID          - デフォルトのTelegramチャット

# 症状：
# KeyError: 'TELEGRAM_BOT_TOKEN'

# 修正：デプロイ時にすべての環境変数が設定されていることを確認
export TELEGRAM_BOT_TOKEN="123456:ABC-..."
export GATEWAY_URL="https://gateway.example.com"
export GATEWAY_TOKEN="gw_..."
export TARGET_CHAT_ID="-1001234567890"

2. トピックIDの不一致

# 症状：
# {'ok': False, 'error_code': 400, 'description': 'Bad Request: chat not found'}

# 原因: message_thread_id（トピック）が存在しない、またはフォーラムが有効でない

# 修正: 対象チャットでトピックが有効になっていることを確認：
# /setname Your Forum Name → Enable Forum
# その後、以下から数値トピックIDを取得：
curl -s "https://api.telegram.org/bot$TOKEN/getForumTopicByChat" \
  -d "chat_id=$CHAT_ID" -d "title=Target Topic"

3. Gatewayセッションが存在しない

# 症状：
# {'ok': False, 'error_code': 404, 'description': 'Session not found'}

# 修正: 連携前にセッションが存在することを確認：
curl -s -H "Authorization: Bearer $GATEWAY_TOKEN" \
  "$GATEWAY_URL/sessions" | jq '.[] | select(.id=="agent-b")'

# または事前にセッションを作成：
curl -s -X POST -H "Authorization: Bearer $GATEWAY_TOKEN" \
  "$GATEWAY_URL/sessions/agent-b" \
  -d '{"config": {"topic_id": 20}}'

4. 確認配信時の競合状態

# 症状: 確認が送信者の次のメッセージを処理した後到着
# （メッセージが順序通りに表示されない視覚的な問題）

# 原因: 順序保証のない非同期配信

# 修正: 肯定応答を含む逐次配信を使用：
result = send_visible_message(topic_id, message)
if result['ok']:
    deliver_payload(session_id, payload)  # 可視メッセージ確定後のみ
    send_confirmation(sender_session, receipt)  # ペイロード配信後のみ

エージェント動作の落とし穴

5. ダブル連携（オリジナル＋フック）

ネイティブフックとエージェントプロンプト命令の両方が実装されている場合、両方が発火する可能性があります：

# シナリオ: ネイティブフックと古いプロンプト命令の両方が発火

# ユーザーが見るメッセージ：
# 📨 Handoff received from Agent A — processing now.  (フックが発火)
# 📨 Handoff received from Agent A — processing now.  (エージェントも発火)

# 修正: ネイティブフックが有効になったらプロンプトレベルの連携命令を削除
# または、エージェントには連携のみを行わせ、メッセージ投稿は行わせない

6. 圧縮時にトピックIDが伝播しない

# 症状: コンテキスト圧縮後、連携先のtopic_idが失われる

# シナリオ：
# エージェントが保持: sessions_to_topic = {'agent-b': 20}
# 圧縮後: sessions_to_topic = {}  (失われた)

# 修正: マッピングをコンテキストではなく永続設定に保存：
# config.yaml:
# handoff_mappings:
#   agent-b:
#     topic_id: 20
#     last_handoff: "2024-01-15T..."

7. 可視メッセージのGatewayタイムアウト

# 症状：
# 30秒後に連携がタイムアウト
# Telegram APIのレスポンスが遅い
# 可視メッセージが投稿されない
# ペイロードは配信されたまま（不整合な状態）

# 修正: リトライ付きのタイムアウトを実装：
def post_with_retry(topic_id, message, max_retries=3):
    for attempt in range(max_retries):
        try:
            return requests.post(
                TELEGRAM_API,
                json={...},
                timeout=10  # ハードタイムアウト
            ).json()
        except requests.Timeout:
            if attempt == max_retries - 1:
                raise HandoffDeliveryError(f"{max_retries}回の試行後に失敗しました")

マルチエージェント設計の落とし穴

8. 循環連携の検出

# 症状: エージェント間の無限連携ループ

# Agent A → Agent B → Agent A → Agent B → ...

# 修正: 連携深度トラッキングを実装：
payload = {
    **payload,
    '_handoff_depth': payload.get('_handoff_depth', 0) + 1,
    '_handoff_chain': [...payload.get('_handoff_chain', []), current_session]
}

if payload['_handoff_depth'] > MAX_HANDOFF_DEPTH:
    raise CircularHandoffError(f"最大深度を超過: {payload['_handoff_chain']}")

9. セッションIDの混乱

# 症状: 配信確認が間違ったセッションに届く
# （他のユーザー向けの確認が表示される）

# 原因: 共有/偽装コンテキストからのsender_session

# 修正: sender_sessionは常に認証済みコンテキストから解決：
sender_session = authenticated_user.session_id  # ペイロードからは取得しない

文脈的に関連する問題

交差参照テーブル

関連するGitHubイシュー

• [FEATURE] sessions_send return receipt — 送信者側の確認に関する以前のリクエスト（クローズ、未実装）
• [BUG] sessions_send silently fails when target session offline — サイレント障害モードが現在の脆弱な回避策の議論を有効にした
• [FEATURE] Event hook system for gateway lifecycle — この機能を包含できる汎用フックアーキテクチャの提案
• [DOCS] Multi-agent handoff patterns documentation — 欠落しているガイダンスにより、各チームがパターンを再発見することを余儀なくされている
• [PERF] sessions_send latency under high load — フックオーバーヘッドはパフォーマンス設計で考慮する必要がある

関連する設定オプション

# この機能と相互作用する可能性のある関連するOpenClaw設定オプション：

{
  "sessions": {
    "handoff_timeout": 30000,        // 連携配信のタイムアウト
    "require_acknowledgment": false,  // 将来: ackまでブロック
    "max_handoff_depth": 5            // 循環連携を防止
  },
  "telegram": {
    "topic_mode": "required",         // トピックの存在を保証
    "rate_limit_per_second": 30       // 自動肯定応答のレートに影響
  },
  "hooks": {
    "sessionSend": { },                // 将来: 送信者側のフック
    "sessionReceive": { }              // この機能
  }
}

外部依存関係

• Telegram Bot API — 可視メッセージの投稿に使用；レート制限と可用性の影響を受ける
• Gateway Sessions API — 配信確認エンドポイントをサポートする必要がある
• Message Queue — 実装する場合、配信順序を保証する必要がある