Message Handlers

export class MessageBusService extends Service {
  static serviceType = "message-bus-service";
  capabilityDescription = "Manages connection and message synchronization with the message server.";

  private boundHandleIncomingMessage: (message: MessageServiceMessage) => Promise<void>;
  private validChannelIds: Set<UUID> = new Set();
  private subscribedServers: Set<UUID> = new Set();
}

private async handleIncomingMessage(message: MessageServiceMessage) {
  // Validate channel participation
  const participants = await this.getChannelParticipants(message.channel_id);
  if (!participants.includes(this.runtime.agentId)) {
    return;
  }

  // Create agent memory from message
  const agentMemory = this.createAgentMemory(message, ...);

  // Emit to runtime for processing
  await this.runtime.emitEvent(EventType.MESSAGE_RECEIVED, {
    runtime: this.runtime,
    message: agentMemory,
    callback: this.handleAgentResponse.bind(this)
  });
}

const messageReceivedHandler = async ({
  runtime,
  message,
  callback,
  onComplete,
}: MessageReceivedHandlerParams): Promise<void> => {
  // Set up timeout monitoring
  const timeoutDuration = 60 * 60 * 1000; // 1 hour
  let timeoutId: NodeJS.Timeout | undefined = undefined;

  try {
    logger.info(`Message received from ${message.entityId} in room ${message.roomId}`);

    // Generate response ID for tracking
    const responseId = v4();

    // Track latest response per agent+room
    const agentResponses = latestResponseIds.get(runtime.agentId) || new Map();
    agentResponses.set(message.roomId, responseId);
    latestResponseIds.set(runtime.agentId, agentResponses);

    // Start run tracking
    const runId = runtime.startRun();
    const startTime = Date.now();

    // Process message with timeout protection
    await processMessage(runtime, message, callback, responseId, runId, startTime);
  } catch (error) {
    logger.error("Message processing failed:", error);
    throw error;
  } finally {
    clearTimeout(timeoutId);
    onComplete?.();
  }
};

// Skip shouldRespond check for certain channel types
const shouldSkipShouldRespond = shouldBypassShouldRespond(runtime, room, message.content.source);

if (!shouldSkipShouldRespond) {
  // Evaluate if agent should respond using customizable template
  const shouldRespondPrompt = composePromptFromState({
    state,
    template: runtime.character.templates?.shouldRespondTemplate || shouldRespondTemplate,
  });

  const response = await runtime.useModel(ModelType.TEXT_SMALL, {
    prompt: shouldRespondPrompt,
  });

  const responseObject = parseKeyValueXml(response);

  // Non-response actions
  const nonResponseActions = ["IGNORE", "NONE"];
  shouldRespond =
    responseObject?.action && !nonResponseActions.includes(responseObject.action.toUpperCase());
}

export const shouldRespondTemplate = `<task>Decide on behalf of {{agentName}} whether they should respond to the message, ignore it or stop the conversation.</task>

<providers>
{{providers}}
</providers>

<instructions>Decide if {{agentName}} should respond to or interact with the conversation.
If the message is directed at or relevant to {{agentName}}, respond with RESPOND action.
If a user asks {{agentName}} to be quiet, respond with STOP action.
If {{agentName}} should ignore the message, respond with IGNORE action.</instructions>

<output>
<response>
  <name>{{agentName}}</name>
  <reasoning>Your reasoning here</reasoning>
  <action>RESPOND | IGNORE | STOP</action>
</response>
</output>`;

export const messageHandlerTemplate = `<task>Generate dialog and actions for the character {{agentName}}.</task>

<providers>
{{providers}}
</providers>

These are the available valid actions:
<actionNames>
{{actionNames}}
</actionNames>

<instructions>
Write a thought and plan for {{agentName}} and decide what actions to take.

IMPORTANT ACTION ORDERING RULES:
- Actions are executed in the ORDER you list them
- REPLY should come FIRST to acknowledge the user's request
- Common patterns:
  - For requests: REPLY,CALL_MCP_TOOL
  - For tasks: REPLY,SEND_MESSAGE
  - For multi-step: REPLY,ACTION1,ACTION2
</instructions>

<output>
<response>
    <thought>Your thought here</thought>
    <actions>ACTION1,ACTION2</actions>
    <providers>PROVIDER1,PROVIDER2</providers>
    <text>Your response text here</text>
</response>
</output>`;

export const postCreationTemplate = `# Task: Create a post in the voice and style of {{agentName}}.

<response>
  <thought>I am thinking about writing a post about [topic]</thought>
  <post>The actual post content</post>
  <imagePrompt>A description for an image to accompany the post</imagePrompt>
</response>`;

export const imageDescriptionTemplate = `<task>Describe the image for {{agentName}}</task>

<output>
<response>
  <title>Short title for the image</title>
  <description>Detailed description of what's in the image</description>
  <text>Natural language description as if explaining to someone</text>
</response>
</output>`;

export const continueTemplate = `<task>Continue the conversation as {{agentName}}</task>

<instructions>
Continue naturally, maintaining character personality and context.
</instructions>`;

export const evaluationTemplate = `<task>Evaluate the following as {{agentName}}</task>

<instructions>
Provide thoughtful evaluation based on the character's knowledge and perspective.
</instructions>`;

{
  "name": "TechSupport",
  "bio": ["Technical support specialist"],
  "templates": {
    "shouldRespondTemplate": "<task>As a tech support agent, determine if this message needs assistance...</task>",
    "messageHandlerTemplate": "<task>Provide technical support as {{agentName}}...</task>",
    "postCreationTemplate": "<task>Create a tech tip post...</task>"
  }
}

// Override templates during runtime
runtime.character.templates = {
  ...runtime.character.templates,
  messageHandlerTemplate: customMessageTemplate,
  shouldRespondTemplate: customShouldRespondTemplate
};

export const customPlugin: Plugin = {
  name: 'custom-templates',
  actions: [],
  providers: [],
  evaluators: [],
  services: [],
  
  // Initialize custom templates
  async onInitialize(runtime: IAgentRuntime) {
    if (!runtime.character.templates) {
      runtime.character.templates = {};
    }
    
    // Set custom templates based on character type
    if (runtime.character.type === 'support') {
      runtime.character.templates.messageHandlerTemplate = supportAgentTemplate;
      runtime.character.templates.shouldRespondTemplate = supportShouldRespondTemplate;
    }
  }
};

export function shouldBypassShouldRespond(
  runtime: IAgentRuntime,
  room?: Room,
  source?: string
): boolean {
  const defaultBypassTypes = [
    ChannelType.DM,         // Direct messages from any platform
    ChannelType.VOICE_DM,   // Voice DMs from any platform
    ChannelType.SELF,       // Self-messages
    ChannelType.API,        // API-initiated messages
  ];

  const defaultBypassSources = [
    "client_chat",   // Web UI chat
    "api",          // REST API
    "postman",      // Postman testing
  ];

  // Check room type and source (platform-agnostic)
  const roomType = room?.type?.toString().toLowerCase();
  const sourceStr = source?.toLowerCase() || "";

  return bypassTypes.has(roomType) || bypassSources.some((pattern) => sourceStr.includes(pattern));
}

if (shouldRespond) {
  // Compose state with action information
  state = await runtime.composeState(message, ["ACTIONS"]);

  const prompt = composePromptFromState({
    state,
    template: runtime.character.templates?.messageHandlerTemplate || messageHandlerTemplate,
  });

  let responseContent: Content | null = null;
  let retries = 0;
  const maxRetries = 3;

  // Retry mechanism for incomplete responses
  while (retries < maxRetries && (!responseContent?.thought || !responseContent?.actions)) {
    let response = await runtime.useModel(ModelType.TEXT_LARGE, {
      prompt,
    });

    // Parse XML response
    const parsedXml = parseKeyValueXml(response);

    if (parsedXml) {
      responseContent = {
        thought: parsedXml.thought || "",
        actions: parsedXml.actions || ["IGNORE"],
        providers: parsedXml.providers || [],
        text: parsedXml.text || "",
        simple: parsedXml.simple || false,
      };
    }

    retries++;
  }
}

if (responseContent && message.id) {
  responseContent.inReplyTo = createUniqueUuid(runtime, message.id);

  // Determine if response is simple
  const isSimple =
    responseContent.actions?.length === 1 &&
    responseContent.actions[0].toUpperCase() === "REPLY" &&
    (!responseContent.providers || responseContent.providers.length === 0);

  responseContent.simple = isSimple;

  const responseMessage = {
    id: asUUID(v4()),
    entityId: runtime.agentId,
    agentId: runtime.agentId,
    content: responseContent,
    roomId: message.roomId,
    createdAt: Date.now(),
  };

  responseMessages = [responseMessage];
}

if (responseContent && responseContent.simple && responseContent.text) {
  // Log provider usage for simple responses
  if (responseContent.providers && responseContent.providers.length > 0) {
    logger.debug("Simple response used providers", responseContent.providers);
  }

  // Execute callback with response content
  await callback(responseContent);
} else {
  // Process complex actions
  await runtime.processActions(message, responseMessages, state, callback);
}

// Runtime processes actions based on responseContent.actions
await runtime.processActions(message, responseMessages, state, callback);

// Common actions include:
// - REPLY: Send text response
// - IGNORE: No response
// - FOLLOW_ROOM: Follow room updates
// - SEND_MESSAGE: Send targeted message
// - UPDATE_ENTITY: Update entity information

export async function processAttachments(
  attachments: Media[],
  runtime: IAgentRuntime
): Promise<Media[]> {
  const processedAttachments: Media[] = [];

  for (const attachment of attachments) {
    const processedAttachment: Media = { ...attachment };

    // Process images
    if (attachment.contentType === ContentType.IMAGE && !attachment.description) {
      const response = await runtime.useModel(ModelType.IMAGE_DESCRIPTION, {
        prompt: imageDescriptionTemplate,
        imageUrl: attachment.url,
      });

      const parsedXml = parseKeyValueXml(response);
      if (parsedXml?.description) {
        processedAttachment.description = parsedXml.description;
        processedAttachment.title = parsedXml.title || "Image";
        processedAttachment.text = parsedXml.text;
      }
    }

    // Process documents
    if (attachment.contentType === ContentType.DOCUMENT && !attachment.text) {
      const res = await fetch(attachment.url);
      const contentType = res.headers.get("content-type") || "";

      if (contentType.startsWith("text/plain")) {
        const textContent = await res.text();
        processedAttachment.text = textContent;
        processedAttachment.title = processedAttachment.title || "Text File";
      }
    }

    processedAttachments.push(processedAttachment);
  }

  return processedAttachments;
}

// Save incoming message to memory
await Promise.all([
  runtime.addEmbeddingToMemory(message),
  runtime.createMemory(message, "messages"),
]);

const latestResponseIds = new Map<string, Map<string, string>>();

// Set latest response ID for agent+room
const agentResponses = latestResponseIds.get(runtime.agentId) || new Map();
agentResponses.set(message.roomId, responseId);

// Check if still latest before processing
const currentResponseId = agentResponses.get(message.roomId);
if (currentResponseId !== responseId) {
  logger.info('Response discarded - newer message being processed');
  return;
}

if (!shouldRespond) {
  // Create ignore content
  const ignoreContent: Content = {
    thought: "Agent decided not to respond to this message.",
    actions: ["IGNORE"],
    simple: true,
    inReplyTo: createUniqueUuid(runtime, message.id),
  };

  // Execute callback with ignore content
  await callback(ignoreContent);

  // Save ignore decision to memory
  const ignoreMemory = {
    id: asUUID(v4()),
    entityId: runtime.agentId,
    agentId: runtime.agentId,
    content: ignoreContent,
    roomId: message.roomId,
    createdAt: Date.now(),
  };

  await runtime.createMemory(ignoreMemory, "messages");
}

const timeoutDuration = 60 * 60 * 1000; // 1 hour
const timeoutPromise = new Promise<never>((_, reject) => {
  timeoutId = setTimeout(async () => {
    await runtime.emitEvent(EventType.RUN_TIMEOUT, {
      runtime,
      runId,
      messageId: message.id,
      roomId: message.roomId,
      entityId: message.entityId,
      startTime,
      status: "timeout",
      endTime: Date.now(),
      duration: Date.now() - startTime,
      error: "Run exceeded 60 minute timeout",
      source: "messageHandler",
    });
    reject(new Error("Run exceeded 60 minute timeout"));
  }, timeoutDuration);
});

await Promise.race([processingPromise, timeoutPromise]);

try {
  await processMessage(runtime, message, callback);
} catch (error) {
  logger.error("Message processing failed:", error);

  // Emit run ended event with error
  await runtime.emitEvent(EventType.RUN_ENDED, {
    runtime,
    runId,
    messageId: message.id,
    roomId: message.roomId,
    entityId: message.entityId,
    startTime,
    status: "error",
    endTime: Date.now(),
    duration: Date.now() - startTime,
    error: error.message,
    source: "messageHandler",
  });
}

const reactionReceivedHandler = async ({
  runtime,
  message,
}: {
  runtime: IAgentRuntime;
  message: Memory;
}) => {
  try {
    await runtime.createMemory(message, "messages");
  } catch (error: any) {
    if (error.code === "23505") {
      logger.warn("Duplicate reaction memory, skipping");
      return;
    }
    logger.error("Error in reaction handler:", error);
  }
};

const messageDeletedHandler = async ({
  runtime,
  message,
}: {
  runtime: IAgentRuntime;
  message: Memory;
}) => {
  try {
    if (!message.id) {
      logger.error("Cannot delete memory: message ID is missing");
      return;
    }

    logger.info("Deleting memory for message", message.id);
    await runtime.deleteMemory(message.id);
  } catch (error) {
    logger.error("Error in message deleted handler:", error);
  }
};

const channelClearedHandler = async ({
  runtime,
  roomId,
  channelId,
  memoryCount,
}: {
  runtime: IAgentRuntime;
  roomId: UUID;
  channelId: string;
  memoryCount: number;
}) => {
  try {
    logger.info(`Clearing ${memoryCount} message memories from channel ${channelId}`);

    const memories = await runtime.getMemoriesByRoomIds({
      tableName: "messages",
      roomIds: [roomId],
    });

    let deletedCount = 0;
    for (const memory of memories) {
      if (memory.id) {
        try {
          await runtime.deleteMemory(memory.id);
          deletedCount++;
        } catch (error) {
          logger.warn(`Failed to delete message memory ${memory.id}:`, error);
        }
      }
    }

    logger.info(`Successfully cleared ${deletedCount}/${memories.length} message memories`);
  } catch (error) {
    logger.error("Error in channel cleared handler:", error);
  }
};

const events = {
  [EventType.MESSAGE_RECEIVED]: [
    async (payload: MessagePayload) => {
      if (!payload.callback) {
        logger.error("No callback provided for message");
        return;
      }
      await messageReceivedHandler({
        runtime: payload.runtime,
        message: payload.message,
        callback: payload.callback,
        onComplete: payload.onComplete,
      });
    },
  ],

  [EventType.VOICE_MESSAGE_RECEIVED]: [
    async (payload: MessagePayload) => {
      // Use same handler for voice messages
      await messageReceivedHandler({
        runtime: payload.runtime,
        message: payload.message,
        callback: payload.callback,
        onComplete: payload.onComplete,
      });
    },
  ],

  [EventType.REACTION_RECEIVED]: [
    async (payload: MessagePayload) => {
      await reactionReceivedHandler({
        runtime: payload.runtime,
        message: payload.message,
      });
    },
  ],

  [EventType.MESSAGE_DELETED]: [
    async (payload: MessagePayload) => {
      await messageDeletedHandler({
        runtime: payload.runtime,
        message: payload.message,
      });
    },
  ],

  [EventType.CHANNEL_CLEARED]: [
    async (payload: EventPayload & { roomId: UUID; channelId: string; memoryCount: number }) => {
      await channelClearedHandler({
        runtime: payload.runtime,
        roomId: payload.roomId,
        channelId: payload.channelId,
        memoryCount: payload.memoryCount,
      });
    },
  ],
};

export const bootstrapPlugin: Plugin = {
  name: "bootstrap",
  description: "Agent bootstrap with basic actions and evaluators",
  events: events as any as PluginEvents,
  // ... other plugin properties
};

// Each agent+room combination tracks its latest response
const latestResponseIds = new Map<string, Map<string, string>>();

// Concurrent safety check
const currentResponseId = agentResponses.get(message.roomId);
if (currentResponseId !== responseId) {
  logger.info('Response discarded - newer message being processed');
  return;
}

// Clean up response tracking
agentResponses.delete(message.roomId);
if (agentResponses.size === 0) {
  latestResponseIds.delete(runtime.agentId);
}