popagent

Latest change 7f6c1d0ed24ffc264e50e9966eaf45024245ba71 - feat: add per-task agent communication by AkurAI Build

import { createTool } from "@mastra/core/tools";
import { z } from "zod";
import {
  type TaskAgentCommunicationMessage,
  type TaskAgentDeliveryReceipt,
  type TaskAgentPeer,
  TaskAgentCommunicationSession,
} from "../task-agent-communication";

const inputSchema = z.object({
  op: z.enum(["list", "send", "inbox", "wait"]),
  to: z.string().min(1).max(128).optional(),
  message: z.string().min(1).max(4_000).optional(),
  replyTo: z.string().uuid().optional(),
  from: z.string().min(1).max(128).optional(),
  timeoutMs: z.number().int().min(0).max(30_000).optional(),
  peek: z.boolean().optional(),
}).strict();

const peerSchema = z.object({
  id: z.string(),
  state: z.enum(["idle", "active", "waiting", "terminal"]),
});
const messageSchema = z.object({
  id: z.string(),
  taskId: z.string(),
  workspaceId: z.string(),
  resourceId: z.string(),
  from: z.string(),
  to: z.string(),
  body: z.string(),
  replyTo: z.string().optional(),
  createdAt: z.string(),
});
const receiptSchema = z.object({
  to: z.string(),
  outcome: z.enum(["delivered", "queued", "failed"]),
  messageId: z.string().optional(),
  error: z.string().optional(),
});
const outputSchema = z.object({
  peers: z.array(peerSchema).optional(),
  receipts: z.array(receiptSchema).optional(),
  messages: z.array(messageSchema).optional(),
  message: messageSchema.nullable().optional(),
});

type ToolOutput = {
  peers?: TaskAgentPeer[];
  receipts?: TaskAgentDeliveryReceipt[];
  messages?: TaskAgentCommunicationMessage[];
  message?: TaskAgentCommunicationMessage | null;
};


export function createAgentCommunicationTool(
  agentId: string,
  session: TaskAgentCommunicationSession,
) {
  return createTool({
    id: "agent-communication",
    description: "Coordinate with independently participating agents in this task. List task peers; send to one peer or to all; consume queued inbox messages once; or wait at a real delivery boundary. A queued receipt means the peer has not consumed the message yet; delivered means an active wait received it.",
    inputSchema,
    outputSchema,
    execute: async (input): Promise<ToolOutput> => {
      if (input.op === "list") return { peers: session.list(agentId) };
      if (input.op === "inbox") return { messages: session.inbox(agentId, { peek: input.peek }) };
      if (input.op === "wait") {
        return { message: await session.wait(agentId, { from: input.from, timeoutMs: input.timeoutMs }) };
      }
      if (!input.to || !input.message) throw new Error("send requires to and message");
      if (input.to === "all") {
        if (input.replyTo) throw new Error("broadcast messages cannot be replies");
        return { receipts: await session.broadcast(agentId, input.message) };
      }
      return {
        receipts: [await session.send({
          from: agentId,
          to: input.to,
          body: input.message,
          replyTo: input.replyTo,
        })],
      };
    },
  });
}