Las llamadas de herramienta de mensaje se agrupan hasta el final del turno en lugar de entregarse inmediatamente - Message Tool Calls Batched Until Turn End Instead of Delivered Immediately

Comportamiento observable

Cuando un agente invoca la herramienta message múltiples veces dentro de un solo turno, todos los mensajes llegan simultáneamente al finalizar el turno en lugar de en sus respectivos puntos de invocación.

Demostración CLI del comportamiento actual

# Escenario: Agente con 3 llamadas a la herramienta message durante una tarea de larga duración
# El usuario experimenta: silencio durante toda la duración, luego todos los mensajes llegan de golpe

# Línea de tiempo de eventos (como se ve por el canal/consumidor de API):
[T+0s]   Turno iniciado - sin salida visible
[T+30s]  Llamadas de herramienta 1-5 ejecutadas - sin salida visible  
[T+60s]  Llamadas de herramienta 6-10 ejecutadas - sin salida visible
[T+90s]  Turno completado - TODOS LOS TRES MENSAJES entregados simultáneamente:

Salida del canal (recibida en T+90s):
┌─────────────────────────────────────────────────────────────┐
│ [90s] 收到，开始分析...                                      │
│ [90s] 数据拉完，正在生成报告                                 │
│ [90s] 报告完成，核心结论...                                  │
└─────────────────────────────────────────────────────────────┘

Salida esperada:
┌─────────────────────────────────────────────────────────────┐
│ [0s]   收到，开始分析...                                     │
│ [60s]  数据拉完，正在生成报告                                 │
│ [90s]  报告完成，核心结论...                                  │
└─────────────────────────────────────────────────────────────┘

Manifestaciones específicas por canal

Indicador de depuración

Cuando el rastreo está habilitado, la salida de la herramienta message muestra comportamiento de agrupamiento:

# Con TRACE_LEVEL=debug, observar el ciclo de vida del turno
[TRACE] Turno 42 iniciado
[TRACE] Llamada a herramienta: message (en cola para entrega al fin del turno) - "收到，开始分析..."
[TRACE] Llamada a herramienta: database.query (ejecutando)
[TRACE] Llamada a herramienta: message (en cola para entrega al fin del turno) - "数据拉完，正在生成报告"
[TRACE] Llamada a herramienta: file.write (ejecutando)
[TRACE] Llamada a herramienta: message (en cola para entrega al fin del turno) - "报告完成，核心结论..."
[TRACE] Turno 42 completado - descargando 3 mensajes en cola
[DEBUG] Entregando lote: [msg_1, msg_2, msg_3]

Contraste con escenarios funcionales

Los mensajes sí llegan inmediatamente cuando:

• Un turno contiene solo una única llamada a la herramienta message sin otras herramientas
• El agente completa un turno (todas las herramientas no-message), luego inicia un nuevo turno con un mensaje
• El mensaje se envía mediante session.reply() en lugar de la herramienta message

Análisis arquitectónico

La falla de entrega inmediata se origina en el modelo de agregación de resultados con ámbito de turno de OpenClaw. El sistema está arquitectado para recolectar todos los resultados de herramientas—including salidas de la herramienta message—dentro de un límite de turno antes de entregarlos en un solo lote.

Desglose del flujo de código

┌──────────────────────────────────────────────────────────────────────────────┐
│                         TURN PROCESSING PIPELINE                             │
├──────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  1. TURN_START                                                               │
│     └─> Inicializar contexto del turno                                       │
│     └─> Crear buffer de resultados vacío                                     │
│                                                                              │
│  2. TOOL_EXECUTION_LOOP                                                      │
│     ├─> Para cada llamada de herramienta:                                    │
│     │   ├─> Ejecutar herramienta                                           │
│     │   ├─> Si herramienta == "message":                                    │
│     │   │   └─> buffer.append(message_result)  ← EN COLA, NO ENVIADO      │
│     │   │                                                             ↑      │
│     │   └─> buffer.append(tool_result)                                     │      │
│     │                                                                     │      │
│     └─> Repetir hasta que no haya más llamadas de herramienta            ─┘      │
│                                                                              │
│  3. TURN_END                                                                 │
│     └─> flush_result_buffer()  ← TODOS LOS MENSAJES ENVIADOS AQUÍ           │
│     └─> deliver_to_channel(batch)                                           │
│                                                                              │
└──────────────────────────────────────────────────────────────────────────────┘

Archivos de código fuente clave involucrados

Desajuste semántico

La herramienta message es semánticamente una notificación de usuario tipo fire-and-forget, sin embargo la implementación la trata idénticamente a otras herramientas que devuelven datos estructurados:

// Implementación actual (problemática)
class MessageTool {
  async execute(params: MessageParams, context: TurnContext): Promise {
    // Trata el mensaje como una herramienta que devuelve datos
    // El resultado se pone en cola en context.results[] hasta el fin del turno
    return {
      output: `Message queued: ${params.content}`,
      // Sin entrega inmediata al canal
    };
  }
}

// Intención semántica
// Herramienta message = "Enviar esto al usuario AHORA"
// Otras herramientas = "Devolver este resultado para consideración del agente"

Comparación con session.reply()

El método session.reply() entrega inmediatamente porque evita el buffer de resultados:

// session.reply() - ruta de entrega inmediata
class Session {
  async reply(content: string): Promise {
    await this.channel.send(content);  // ← Envío directo al canal
  }
}

// herramienta message - ruta de entrega diferida  
class MessageTool {
  async execute(params, context): Promise {
    context.results.push({ output: content });  // ← En buffer
    // Entregado solo cuando el turno completa
  }
}

Por qué existe este diseño

El modelo de agrupamiento sirve casos de uso válidos:

• Reduce llamadas API a canales (un lote vs muchos envíos individuales)
• Asegura el orden de mensajes en relación con los resultados de herramientas
• Simplifica las implementaciones de canales (una sola respuesta por turno)

Sin embargo, entra en conflicto con la intención semántica de una herramienta “enviar mensaje al usuario”, que implica inmediatez.

Solución recomendada: Híbrido de Opción A + Opción C

Implementar entrega inmediata por defecto para la herramienta message mientras se proporciona una bandera immediate: false para casos que requieren entrega agrupada.

Fase 1: Modificar el esquema de la herramienta Message

Archivo: packages/tools/message/src/schema.ts

// ANTES
export const messageToolSchema = {
  name: "message",
  description: "Send a message to the user",
  parameters: {
    type: "object",
    properties: {
      content: {
        type: "string",
        description: "The message content to send to the user"
      }
    },
    required: ["content"]
  }
};

// DESPUÉS
export const messageToolSchema = {
  name: "message",
  description: "Send a message to the user. Messages are delivered immediately unless batch mode is requested.",
  parameters: {
    type: "object",
    properties: {
      content: {
        type: "string", 
        description: "The message content to send to the user"
      },
      immediate: {
        type: "boolean",
        description: "If true, deliver immediately. If false, queue until turn end. Defaults to true.",
        default: true
      }
    },
    required: ["content"]
  }
};

Fase 2: Actualizar la implementación de la herramienta Message

Archivo: packages/tools/message/src/message-tool.ts

import { Tool, ToolResult, TurnContext } from "@openclaw/core";
import { channelRegistry } from "@openclaw/channel-core";

interface MessageParams {
  content: string;
  immediate?: boolean;
}

// Track messages that should be delivered immediately
const IMMEDIATE_DELIVERY_THRESHOLD_MS = 0; // 0 = always immediate when requested

export class MessageTool implements Tool {
  name = "message";
  description = messageToolSchema.description;
  parameters = messageToolSchema.parameters;

  async execute(
    params: MessageParams,
    context: TurnContext
  ): Promise {
    const content = params.content;
    const shouldDeliverImmediately = params.immediate !== false; // Default: true

    if (shouldDeliverImmediately) {
      // RUTA DE ENTREGA INMEDIATA
      return this.deliverImmediately(content, context);
    } else {
      // RUTA DE ENTREGA AGRUPADA (comportamiento original)
      return this.queueForTurnEnd(content, context);
    }
  }

  private async deliverImmediately(
    content: string,
    context: TurnContext
  ): Promise {
    try {
      // Obtener el canal activo para esta sesión
      const channel = channelRegistry.getChannel(context.session.channelType);
      
      // Enviar directamente al canal, fuera del buffer del turno
      await channel.send({
        sessionId: context.session.id,
        content: content,
        metadata: {
          toolName: "message",
          deliveredAt: Date.now(),
          deliveryMode: "immediate"
        }
      });

      return {
        success: true,
        output: `Message delivered immediately: ${content.substring(0, 50)}...`,
        metadata: {
          deliveredAt: Date.now(),
          deliveryMode: "immediate"
        }
      };
    } catch (error) {
      return {
        success: false,
        output: "",
        error: `Failed to deliver message immediately: ${error.message}`,
        metadata: {
          deliveryMode: "immediate",
          fellBackToBatch: true
        }
      };
    }
  }

  private async queueForTurnEnd(
    content: string,
    context: TurnContext
  ): Promise {
    // Comportamiento original: agregar al buffer del turno
    context.results.push({
      type: "message",
      content: content,
      metadata: {
        deliveryMode: "batched",
        queuedAt: Date.now()
      }
    });

    return {
      success: true,
      output: `Message queued for turn-end delivery: ${content.substring(0, 50)}...`,
      metadata: {
        deliveryMode: "batched"
      }
    };
  }
}

Fase 3: Registrar capacidad de envío del canal

Archivo: packages/channel-core/src/channel-registry.ts

// Asegurar que los canales implementen capacidad de envío inmediato
export interface ChannelAdapter {
  // Métodos existentes...
  sendBatch(results: TurnResult[]): Promise;
  
  // NUEVO: Envío inmediato de mensaje único
  send(params: {
    sessionId: string;
    content: string;
    metadata?: Record;
  }): Promise;
}

Fase 4: Actualizar Turn Executor (Cambio mínimo)

Archivo: packages/core/src/turn/turn-executor.ts

// Agregar filtro para excluir mensajes ya entregados del lote
async function flushResults(context: TurnContext): Promise {
  // Filtrar mensajes que fueron entregados inmediatamente
  const batchableResults = context.results.filter(
    result => result.metadata?.deliveryMode !== "immediate"
  );

  if (batchableResults.length > 0) {
    await context.channel.sendBatch(batchableResults);
  }
  
  // Registrar resumen
  const immediateCount = context.results.filter(
    r => r.metadata?.deliveryMode === "immediate"
  ).length;
  
  if (immediateCount > 0) {
    context.logger.debug(`Delivered ${immediateCount} messages immediately`);
  }
}

Fase 5: Opción de configuración

Archivo: packages/core/src/config/tool-config.ts

export interface ToolConfig {
  message: {
    // Modo de entrega por defecto para la herramienta message
    defaultDeliveryMode: "immediate" | "batched";
    // Respaldo si el canal no soporta entrega inmediata
    fallbackToBatchOnError: boolean;
  };
}

export const defaultToolConfig: ToolConfig = {
  message: {
    defaultDeliveryMode: "immediate",  // Cambiado de "batched"
    fallbackToBatchOnError: true
  }
};

Verificación de los cambios

Después de la implementación, el flujo de ejecución se convierte en:

┌──────────────────────────────────────────────────────────────────────────────┐
│                         UPDATED PIPELINE (with fix)                          │
├──────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  1. TURN_START                                                               │
│     └─> Inicializar contexto del turno                                       │
│                                                                              │
│  2. TOOL_EXECUTION_LOOP                                                      │
│     ├─> Llamada a herramienta: message ("收到，开始分析...")                  │
│     │   └─> channel.send() ← ENTREGA INMEDIATA                              │
│     │   └─> return { deliveredAt, deliveryMode: "immediate" }               │
│     │                                                                     │
│     ├─> Llamada a herramienta: database.query                               │
│     │   └─> context.results.push(result)  ← Buffer normal                  │
│     │                                                                     │
│     ├─> Llamada a herramienta: message ("数据拉完...")                      │
│     │   └─> channel.send() ← ENTREGA INMEDIATA                              │
│     │                                                                     │
│     └─> Llamada a herramienta: file.write                                   │
│         └─> context.results.push(result)                                     │
│                                                                              │
│  3. TURN_END                                                                 │
│     └─> flushResults() - solo resultados no-inmediatos                      │
│     └─> channel.sendBatch([query_result, write_result])                     │
│                                                                              │
└──────────────────────────────────────────────────────────────────────────────┘

Caso de prueba 1: Verificación de entrega inmediata

Propósito: Confirmar que los mensajes llegan en el momento de la invocación, no al fin del turno.

# Script de prueba: verificar tiempo de mensaje
#!/bin/bash

START_TIME=$(date +%s.%N)

# Invocar agente con llamadas a herramienta message temporizadas
curl -X POST http://localhost:3000/api/sessions/test-001/invoke \
  -H "Content-Type: application/json" \
  -d '{
    "message": "Perform 3 searches and send progress after each"
  }'

# Capturar tiempos de entrega de mensajes desde logs del canal
# Esperado: 3 marcas de tiempo de entrega separadas
# Actual (antes del fix): marca de tiempo única al fin del turno

echo "Checking message delivery timestamps..."
grep "Message delivered" /var/log/openclaw/channel.log | \
  awk '{print $1, $2, $8}' | \
  sort -u

Salida esperada (después del fix):

2024-01-15 10:30:00.123 deliveredAt=1705315800123
2024-01-15 10:30:35.456 deliveredAt=1705315835456  
2024-01-15 10:31:05.789 deliveredAt=1705315865789
2024-01-15 10:31:35.000 TURN_END

Indicador de falla (antes del fix):

2024-01-15 10:31:35.000 deliveredAt=1705315895000  ← Los tres
2024-01-15 10:31:35.000 deliveredAt=1705315895000  ← Misma marca de tiempo
2024-01-15 10:31:35.000 deliveredAt=1705315895000  ← Fin del turno

Caso de prueba 2: Modos de entrega mixtos

Propósito: Verificar que immediate: false aún pone mensajes en cola correctamente.

# Prompt de agente demostrando modos mixtos:
# Usar entrega inmediata para progreso: "Starting task..."
# Usar agrupado para auditoría: "Query executed at X"

# Verificar que mensajes agrupados no aparecen hasta fin del turno
# mientras que mensajes inmediatos sí

# Paso 1: Iniciar monitoreo
tail -f /var/log/openclaw/channel.log | grep -E "(delivered|queued)" &

# Paso 2: Invocar turno con ambos modos
curl -X POST http://localhost:3000/api/sessions/test-002/invoke \
  -d '{"message": "process with both message modes"}'

# Paso 3: Verificar salida
# Debería ver mensajes inmediatos registrados durante ejecución
# Debería ver mensajes agrupados solo en el marcador TURN_END

Caso de prueba 3: Respaldo de compatibilidad del canal

Propósito: Verificar respaldo elegante cuando el canal carece de capacidad de envío inmediato.

# Si channel.send() lanza "Method not implemented",
# verificar que el mensaje recurre al lote en cola

# Probar con canal mock que no implementa send()
const mockChannel = {
  sendBatch: async (results) => { /* existente */ },
  // send() intencionalmente omitido
};

# Invocar herramienta message
# Esperado: succeed vía respaldo, registrado como "deliveredAt: batched"

grep "fellBackToBatch" /var/log/openclaw/tools.log
# Debería mostrar: herramienta message recurrió a modo agrupado

Suite de pruebas de integración

# packages/tools/message/src/__tests__/message-delivery.test.ts

describe("Message Tool Delivery Modes", () => {
  let mockContext: TurnContext;
  let mockChannel: jest.Mocked;
  
  beforeEach(() => {
    mockChannel = {
      send: jest.fn().mockResolvedValue(undefined),
      sendBatch: jest.fn().mockResolvedValue(undefined),
      // ... otros métodos
    };
    
    mockContext = createMockContext({
      channel: mockChannel,
      session: { id: "test-session", channelType: "telegram" }
    });
  });

  test("entrega inmediatamente por defecto", async () => {
    const tool = new MessageTool();
    await tool.execute({ content: "Mensaje inmediato" }, mockContext);
    
    expect(mockChannel.send).toHaveBeenCalledTimes(1);
    expect(mockChannel.send).toHaveBeenCalledWith(
      expect.objectContaining({
        content: "Mensaje inmediato",
        metadata: expect.objectContaining({
          deliveryMode: "immediate"
        })
      })
    );
    expect(mockChannel.sendBatch).not.toHaveBeenCalled();
  });

  test("pone en cola cuando immediate: false", async () => {
    const tool = new MessageTool();
    await tool.execute(
      { content: "Mensaje agrupado", immediate: false },
      mockContext
    );
    
    expect(mockChannel.send).not.toHaveBeenCalled();
    expect(mockContext.results).toContainEqual(
      expect.objectContaining({
        type: "message",
        content: "Mensaje agrupado",
        metadata: { deliveryMode: "batched" }
      })
    );
  });

  test("recurre a lote cuando channel.send() no disponible", async () => {
    mockChannel.send = undefined; // Simular canal no soportado
    
    const tool = new MessageTool();
    const result = await tool.execute(
      { content: "Prueba" },
      mockContext
    );
    
    expect(result.metadata.fellBackToBatch).toBe(true);
    expect(mockContext.results).toContainEqual(
      expect.objectContaining({
        type: "message",
        metadata: { deliveryMode: "batched" }
      })
    );
  });
});

Lista de verificación de verificación manual

• Logs de trace muestran entrega inmediata: grep "deliverImmediately\|Message delivered" logs/trace.log
• Lote de fin de turno excluye mensajes inmediatos: grep "sendBatch" logs/trace.log | jq '.messages | length' debería igualar total de herramientas menos herramientas message
• Separación de tiempo visible: Las marcas de tiempo de entrega de mensajes difieren de la marca de tiempo de fin de turno
• Cambio de configuración respetado: Configurar defaultDeliveryMode: "batched" revierte al comportamiento anterior

Error común 1: Limitación de tasa del canal

Problema: Los envíos inmediatos rápidos pueden activar límites de tasa del canal (ej., Telegram tiene límite de ~30 mensajes/segundo).

Mitigación:

// Implementar regulación para entrega inmediata
class ThrottledChannelAdapter implements ChannelAdapter {
  private sendQueue: Promise = Promise.resolve();
  private minIntervalMs = 100; // Máx 10 mensajes/segundo

  async send(params: SendParams): Promise {
    this.sendQueue = this.sendQueue.then(async () => {
      await this.throttle();
      return this.channel.send(params);
    });
    await this.sendQueue;
  }

  private async throttle(): Promise {
    // Aplicación de límite de tasa
  }
}

Error común 2: Violaciones de orden de mensajes

Problema: Los mensajes inmediatos pueden llegar antes que mensajes agrupados anteriores, rompiendo el orden cronológico.

Escenario:

Secuencia de herramientas:
1. message "Paso 1" (inmediato)     → llega en T+5s
2. database.query (agrupado)        → en cola
3. message "Paso 2" (inmediato)     → llega en T+10s  
4. Fin del turno                     → resultados agrupados llegan en T+15s

El usuario ve:
[T+5s]   Paso 1
[T+10s]  Paso 2
[T+15s]  Resultado de consulta (¿debería haber estado antes del Paso 2?)

Mitigación: Documentar expectativas de orden; los agentes deben usar modos de entrega consistentes para mensajes relacionados.

Error común 3: Sincronización de estado de sesión

Problema: Los mensajes inmediatos pueden referenciar datos que aún no se han comprometido con el estado de sesión.

Ejemplo:

// Flujo de agente que causa inconsistencia
1. message "Starting query for user ${session.userId}"  // inmediato
2. session.set("userId", "123")                          // en cola
3. Fin del turno → estado comprometido

El usuario ve mensaje con userId indefinido (condición de carrera)

Mitigación: Asegurar que las actualizaciones de estado de sesión sean síncronas; diferir escrituras de estado hasta después de que los mensajes inmediatos sean seguros.

Error común 4: Matriz de compatibilidad de adaptadores de canal

Riesgo: No todos los canales soportan envío inmediato; algunos solo soportan respuestas agrupadas.

Acción: Verificar ChannelAdapterCapabilities antes de usar modo inmediato.

Error común 5: Complejidad de trace/logging

Problema: El rastreo se vuelve más complejo con entregas inmediatas y agrupadas entrelazadas.

Mitigación: Incluir deliveryMode y turnId en todas las entradas de log para filtrado:

{
  "timestamp": "...",
  "level": "debug",
  "message": "Message delivered",
  "turnId": 42,
  "deliveryMode": "immediate",
  "sequenceInTurn": 1,
  "content": "收到，开始分析..."
}

Error común 6: Regresión de compatibilidad hacia atrás

Riesgo: Los agentes existentes que dependen del comportamiento agrupado pueden romperse.

Escenarios:

• Agentes que crean mensajes esperando que se agrupen con resultados de herramientas
• UI esperando exactamente N mensajes al fin del turno

Mitigación:

• Por defecto immediate: true pero documentar el cambio prominentemente
• Proporcionar bandera de configuración tool.message.defaultDeliveryMode: "batched" para desactivarla
• Lanzar como característica opt-in primero, luego cambiar el valor por defecto en la siguiente versión mayor

Error común 7: Pruebas en CI/CD

Problema: Las pruebas basadas en tiempo son frágiles en entornos CI con asignación variable de recursos.

Mitigación:

// Usar prueba determinista con tiempo simulado
test("entrega inmediatamente según bandera, no tiempo", async () => {
  const tool = new MessageTool();
  
  await tool.execute({ content: "prueba" }, mockContext);
  
  // Verificar que send() fue llamado (inmediato) o en cola (agrupado)
  // NO: await waitFor(() => sendCalled())
  // SÍ: expect(sendCalled).toBe(true)
});

Issues de GitHub relacionados

Opciones de configuración relacionadas

Códigos de error relacionados

Contexto histórico

Justificación original del agrupamiento (de #18901):

“Agrupar reduce llamadas API y asegura el orden de mensajes. Sin agrupar, un turno con 10 llamadas de herramienta y 5 mensajes resultaría en 15 llamadas API separadas.”

Contraargumento (de discusión del issue #25463):

“La herramienta message semánticamente significa ’entregar al usuario ahora’. Agrupar contradice la intención y rompe casos de uso en tiempo real como actualizaciones de progreso.”

Ruta de resolución: Este fix implementa Opción A (inmediato por defecto) con Opción C (bandera explícita), reconciliando ambas posiciones al hacer la entrega inmediata el valor por defecto mientras preserva el agrupamiento como opt-in para casos de uso específicos.

Referencias cruzadas de documentación

• Referencia de herramienta Message — Esquema actualizado con parámetro immediate
• Arquitectura de procesamiento de turnos — Diagrama de flujo actualizado mostrando ruta de entrega inmediata
• Guía del adaptador de canal — Requisito del método send() para soporte de entrega inmediata
• Guía de migración: v2.x a v3.0 — Aviso de cambio incompatible para el modo de entrega por defecto