[功能请求：为 sessions_send 添加事件钩子以确保交接可见性] - Feature Request: sessions_send Event Hook for Guaranteed Handoff Visibility

当前限制的表现形式

当智能体使用 sessions_send 执行跨会话交接时，通常会观察到以下失败模式：

无确认的静默处理

# Agent A 发起交接
sessions_send(
    target_session="project-alpha",
    payload={"task": "review_pr_442", "priority": "high"}
)
# Agent B 静默处理 — 目标主题中无可见消息
# 用户看不到任何交接发生的提示

脆弱的提示词级别强制执行

当前的缓解措施依赖于经常失效的提示词指令：

# 在 Agent A 的提示词中（压缩或高负载时常被忽略）：
# "调用 sessions_send 后，必须在目标主题发布一条可见消息。"

# 实际情况：Agent 可能压缩上下文，失去此指令，或：
# 1. 调用 sessions_send ✓
# 2. 忘记发布可见消息 ✗
# 3. 用户无法了解交接状态

向发送方无投递确认

# Agent A 发送交接
result = sessions_send(...)

# result 未提供任何确认：
# - 消息已加入队列
# - 目标会话存在
# - 已尝试投递
# Agent A 在不确定状态下运行

用户层面的症状

• 用户无法确定跨会话交接是否成功
• 跨智能体的工作分配没有审计跟踪
• 多智能体工作流显得不透明且不可信
• 调试交接失败需要手动检查日志

架构缺口分析

当前的 sessions_send 实现存在根本性的设计限制：

1. 即发即忘的消息投递

# 当前实现（概念性）
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 │─────▶│ (传输层) │─────▶│ (处理中) │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ │ │ ▼ ▼ ▼ ┌─────────────────┐ (无回执) (无钩子点) │ Telegram Bot │ │ (表现层) │ └─────────────────┘

3. 提示词级别强制执行本质上是不可靠的

当前缓解措施依赖于无法保证的 LLM 行为：

• 上下文窗口压力导致指令丢失
• 压缩过程可能剥离关键行为提示词
• 智能体自主性意味着提示词只是建议而非约束
• sessions_send 与消息发布之间无原子性

4. 缺乏跨会话协调的事件系统

缺少事件钩子意味着：

• Gateway 无法在投递时触发副作用
• 没有机会向发送方返回投递确认
• 自动确认需要手动实现智能体
• 没有声明式配置路径

拟议实现：sessionReceive 事件钩子

步骤 1：在 Gateway 设置中配置钩子

将以下内容添加到您的 OpenClaw gateway 配置中：

{
  "hooks": {
    "sessionReceive": {
      "autoAcknowledge": {
        "enabled": true,
        "message": "📨 Handoff received from {sender} — processing now.",
        "channel": "last",
        "topic_id": "{incoming_topic}"
      },
      "deliveryConfirmation": {
        "enabled": true,
        "confirm_to_session": "{sender_session}",
        "confirm_message": "✅ Delivered to {target_session} at {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`：发送投递回执的会话 ID
• `confirm_message`：回执模板消息

步骤 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"📨 Handoff incoming from {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):
        """向发送方会话发送投递回执。"""
        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：验证发送方主题中的投递确认

检查发送方的会话/主题中的投递回执：

# 发送方主题中的预期消息：
✅ 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'交接 {i} 失败'

print('链验证：全部通过')
"

环境与配置陷阱

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 您的论坛名称 → 启用论坛
# 然后使用以下方式获取数字主题 ID：
curl -s "https://api.telegram.org/bot$TOKEN/getForumTopicByChat" \
  -d "chat_id=$CHAT_ID" -d "title=目标主题"

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 在压缩过程中未传播

# 症状：上下文压缩后，交接的主题 ID 丢失

# 场景：
# Agent 有：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. 会话身份混淆

# 症状：投递确认发到了错误的会话
# （用户看到的是发给另一个用户的确认）

# 原因：sender_session 来自共享/模拟上下文

# 修复：始终从认证上下文解析 sender_session：
sender_session = authenticated_user.session_id  # 不从有效载荷获取

上下文相关的关联问题

交叉引用表

相关 GitHub 问题

• [功能请求] sessions_send 返回回执 — 早期请求发送方确认（已关闭，未实现）
• [Bug] sessions_send 在目标会话离线时静默失败 — 静默失败模式促成了当前脆弱的变通方案讨论
• [功能请求] Gateway 生命周期的事件钩子系统 — 提议的可包含此功能的通用钩子架构
• [文档] 多智能体交接模式文档 — 缺失的指导迫使每个团队重新发现模式
• [性能] 高负载下 sessions_send 延迟 — 钩子开销必须在性能设计中考虑

相关配置选项

# 可能与此功能交互的 OpenClaw 配置选项：

{
  "sessions": {
    "handoff_timeout": 30000,        // 交接投递超时
    "require_acknowledgment": false,  // 未来：阻塞直到确认
    "max_handoff_depth": 5            // 防止循环交接
  },
  "telegram": {
    "topic_mode": "required",         // 确保主题存在
    "rate_limit_per_second": 30       // 影响自动确认速率
  },
  "hooks": {
    "sessionSend": { },                // 未来：发送方钩子
    "sessionReceive": { }              // 此功能
  }
}

外部依赖

• Telegram Bot API — 用于发布可见消息；受速率限制和可用性约束
• Gateway Sessions API — 必须支持投递确认端点
• 消息队列 — 如果实现，必须保证投递顺序