popagent

Latest change 98a9a832366ede73ff5e61d826d68af4f775db92 - Add safe autonomous deployment and complete project guidance by AkurAI Build

import { describe, expect, test } from "bun:test";
import { ChannelEventBus } from "./channel-events";

describe("ChannelEventBus", () => {
  test("isolates channel subscribers and stops after unsubscribe", () => {
    const bus = new ChannelEventBus();
    const received: string[] = [];
    const unsubscribe = bus.subscribe("general", (event) => received.push(`${event.type}:${event.messageId}`));
    bus.message("other", "ignored", null);
    const published = bus.message("general", "message-1", "task-1");
    unsubscribe();
    bus.task("general", "task-1", "running");
    expect(received).toEqual(["message:message-1"]);
    expect(published).toMatchObject({ channelId: "general", taskId: "task-1", taskStatus: null });
  });

  test("streams heartbeats so idle connections remain observable", async () => {
    const bus = new ChannelEventBus(5);
    const controller = new AbortController();
    const reader = bus.stream("general", controller.signal).body!.getReader();
    await reader.read();
    const heartbeat = new TextDecoder().decode((await reader.read()).value);
    expect(heartbeat).toContain(": heartbeat");
    controller.abort();
  });

  test("streams typed server-sent events", async () => {
    const bus = new ChannelEventBus();
    const controller = new AbortController();
    const response = bus.stream("general", controller.signal);
    const reader = response.body!.getReader();
    await reader.read();
    bus.task("general", "task-1", "completed");
    const chunk = new TextDecoder().decode((await reader.read()).value);
    expect(response.headers.get("content-type")).toContain("text/event-stream");
    expect(chunk).toContain("event: channel");
    expect(chunk).toContain('"taskStatus":"completed"');
    controller.abort();
  });
});