popagent

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

import { describe, expect, test } from "bun:test";
import {
  TaskAgentCommunicationSession,
  type TaskAgentCommunicationMessage,
} from "./task-agent-communication";

function session(options: {
  mailboxCap?: number;
  signal?: AbortSignal;
  persisted?: TaskAgentCommunicationMessage[];
} = {}) {
  return new TaskAgentCommunicationSession({
    taskId: "task-1",
    workspaceId: "workspace-1",
    resourceId: "popagent-user",
    participants: ["orchistrator", "researcher", "implementer", "reviewer"],
    mailboxCap: options.mailboxCap,
    signal: options.signal,
    persist: options.persisted
      ? async (message) => { options.persisted!.push(message); }
      : undefined,
  });
}

describe("TaskAgentCommunicationSession", () => {
  test("lists task peers and exchanges a consumed-once reply", async () => {
    const bus = session();
    expect(bus.list("researcher")).toEqual([
      { id: "orchistrator", state: "idle" },
      { id: "implementer", state: "idle" },
      { id: "reviewer", state: "idle" },
    ]);

    const waiting = bus.wait("reviewer", { from: "researcher", timeoutMs: 1_000 });
    const sent = await bus.send({ from: "researcher", to: "reviewer", body: "Please inspect the boundary." });
    expect(sent).toMatchObject({ to: "reviewer", outcome: "delivered" });
    const received = await waiting;
    expect(received).toMatchObject({ from: "researcher", to: "reviewer", body: "Please inspect the boundary." });
    expect(bus.inbox("reviewer")).toEqual([]);

    const reply = await bus.send({
      from: "reviewer",
      to: "researcher",
      body: "The boundary is contained.",
      replyTo: received!.id,
    });
    expect(reply).toMatchObject({ to: "researcher", outcome: "queued" });
    expect(bus.inbox("researcher")).toEqual([
      expect.objectContaining({ from: "reviewer", replyTo: received!.id }),
    ]);
    expect(bus.inbox("researcher")).toEqual([]);
  });

  test("broadcasts once per peer and deterministically drops the oldest queued message", async () => {
    const bus = session({ mailboxCap: 2 });
    const receipts = await bus.broadcast("orchistrator", "Status check");
    expect(receipts).toEqual([
      expect.objectContaining({ to: "researcher", outcome: "queued" }),
      expect.objectContaining({ to: "implementer", outcome: "queued" }),
      expect.objectContaining({ to: "reviewer", outcome: "queued" }),
    ]);
    expect(bus.inbox("reviewer")).toHaveLength(1);

    await bus.send({ from: "orchistrator", to: "researcher", body: "first" });
    await bus.send({ from: "orchistrator", to: "researcher", body: "second" });
    await bus.send({ from: "orchistrator", to: "researcher", body: "third" });
    expect(bus.inbox("researcher").map((message) => message.body)).toEqual(["second", "third"]);
  });

  test("rejects unknown and terminal recipients and persists only successful sends", async () => {
    const persisted: TaskAgentCommunicationMessage[] = [];
    const bus = session({ persisted });
    expect(await bus.send({ from: "researcher", to: "missing", body: "hello" })).toMatchObject({
      outcome: "failed",
      error: expect.stringContaining("Unknown"),
    });
    bus.terminal("reviewer");
    expect(await bus.send({ from: "researcher", to: "reviewer", body: "hello" })).toMatchObject({
      outcome: "failed",
      error: expect.stringContaining("terminal"),
    });
    const receipt = await bus.send({ from: "researcher", to: "implementer", body: "safe" });
    expect(persisted).toEqual([
      expect.objectContaining({
        id: receipt.messageId,
        taskId: "task-1",
        workspaceId: "workspace-1",
        resourceId: "popagent-user",
        from: "researcher",
        to: "implementer",
      }),
    ]);
  });

  test("abort closes waiters, clears mailboxes, and rejects future sends", async () => {
    const controller = new AbortController();
    const bus = session({ signal: controller.signal });
    await bus.send({ from: "researcher", to: "reviewer", body: "queued" });
    const waiting = bus.wait("implementer", { timeoutMs: 5_000 });
    controller.abort(new Error("task cancelled"));
    await expect(waiting).rejects.toThrow("task cancelled");
    expect(bus.inbox("reviewer")).toEqual([]);
    expect(await bus.send({ from: "researcher", to: "reviewer", body: "late" })).toMatchObject({
      outcome: "failed",
      error: expect.stringContaining("closed"),
    });
  });
});