popagent

Latest change f6fedafe4a36ed6468d5b1b819bbcdcbed5a0d53 - Address a peer by display capitalization, and correct the 9Router runbook 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("addresses a participant by its display capitalization", async () => {
    const persisted: TaskAgentCommunicationMessage[] = [];
    const bus = session({ persisted });

    // Agents see the display name "Orchistrator" in the channel and use it verbatim.
    const receipt = await bus.send({ from: "implementer", to: "Orchistrator", body: "Findings ready" });

    expect(receipt.outcome).not.toBe("failed");
    expect(receipt.to).toBe("orchistrator");
    expect(persisted[0]?.to).toBe("orchistrator");
    expect(bus.inbox("orchistrator").map((message) => message.body)).toEqual(["Findings ready"]);
  });


  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"),
    });
  });
  test("wakes an idle addressed participant and accepts a reply only from that recipient", async () => {
    let bus!: TaskAgentCommunicationSession;
    bus = new TaskAgentCommunicationSession({
      taskId: "task-1",
      workspaceId: "workspace-1",
      resourceId: "popagent-user",
      participants: ["implementer", "reviewer"],
      wake: async (recipientId, incoming) => {
        expect(recipientId).toBe("reviewer");
        await bus.send({
          from: recipientId,
          to: incoming.from,
          body: "Reviewed by the addressed participant.",
          replyTo: incoming.id,
        });
      },
    });

    const sent = await bus.send({
      from: "implementer",
      to: "reviewer",
      body: "Please review this.",
    });
    expect(sent).toMatchObject({ outcome: "woken" });
    await bus.settle();
    expect(bus.inbox("implementer")).toEqual([
      expect.objectContaining({
        from: "reviewer",
        to: "implementer",
        replyTo: sent.messageId,
      }),
    ]);
    expect(await bus.send({
      from: "implementer",
      to: "reviewer",
      body: "Pretending to be the reviewer.",
      replyTo: sent.messageId,
    })).toMatchObject({
      outcome: "failed",
      error: expect.stringContaining("recipient"),
    });
  });
  test("queues replies without waking the original sender into acknowledgement loops", async () => {
    const wakes: string[] = [];
    const bus = new TaskAgentCommunicationSession({
      taskId: "task-1",
      workspaceId: "workspace-1",
      resourceId: "popagent-user",
      participants: ["implementer", "reviewer"],
      wake: async (recipientId) => { wakes.push(recipientId); },
    });
    bus.idle("implementer");
    const request = await bus.send({
      from: "implementer",
      to: "reviewer",
      body: "Please review this.",
    });
    bus.idle("implementer");
    expect(await bus.send({
      from: "reviewer",
      to: "implementer",
      body: "Review complete.",
      replyTo: request.messageId,
    })).toMatchObject({ outcome: "queued" });
    await bus.settle();
    expect(wakes).toEqual(["reviewer"]);
  });
});