popagent

Latest change 74d33974c7d2e99d005596af3bce84f97c7b5f56 - Fix autonomous scheduling and runtime panel behavior by AkurAI Build

import type { AgentTaskStatus, ChannelEvent } from "./api-types";

type Listener = (event: ChannelEvent) => void;

export class ChannelEventBus {
  private readonly listeners = new Map<string, Set<Listener>>();

  constructor(private readonly heartbeatMs = 15_000) {}

  publish(channelId: string, event: Omit<ChannelEvent, "id" | "channelId" | "createdAt">): ChannelEvent {
    const published: ChannelEvent = {
      ...event,
      id: crypto.randomUUID(),
      channelId,
      createdAt: new Date().toISOString(),
    };
    for (const listener of this.listeners.get(channelId) ?? []) listener(published);
    return published;
  }

  message(channelId: string, messageId: string, taskId: string | null) {
    return this.publish(channelId, { type: "message", messageId, taskId, taskStatus: null });
  }

  task(channelId: string, taskId: string, taskStatus: AgentTaskStatus) {
    return this.publish(channelId, { type: "task", messageId: null, taskId, taskStatus });
  }

  subscribe(channelId: string, listener: Listener): () => void {
    const channelListeners = this.listeners.get(channelId) ?? new Set<Listener>();
    channelListeners.add(listener);
    this.listeners.set(channelId, channelListeners);
    return () => {
      channelListeners.delete(listener);
      if (!channelListeners.size) this.listeners.delete(channelId);
    };
  }

  stream(channelId: string, signal?: AbortSignal): Response {
    const encoder = new TextEncoder();
    let unsubscribe: () => void = () => undefined;
    let heartbeat: Timer | undefined;
    let closed = false;
    const cleanup = () => {
      if (closed) return;
      closed = true;
      if (heartbeat) clearInterval(heartbeat);
      unsubscribe();
    };
    const stream = new ReadableStream<Uint8Array>({
      start: (controller) => {
        if (signal?.aborted) {
          cleanup();
          try { controller.close(); } catch {}
          return;
        }
        const send = (event: ChannelEvent) => {
          if (!closed) controller.enqueue(encoder.encode(`id: ${event.id}\nevent: channel\ndata: ${JSON.stringify(event)}\n\n`));
        };
        unsubscribe = this.subscribe(channelId, send);
        controller.enqueue(encoder.encode(": connected\n\n"));
        heartbeat = setInterval(() => {
          if (!closed) controller.enqueue(encoder.encode(": heartbeat\n\n"));
        }, this.heartbeatMs);
        signal?.addEventListener("abort", () => {
          cleanup();
          try { controller.close(); } catch {}
        }, { once: true });
      },
      cancel: () => {
        cleanup();
      },
    });
    return new Response(stream, {
      headers: {
        "content-type": "text/event-stream; charset=utf-8",
        "cache-control": "no-cache, no-transform",
        "x-accel-buffering": "no",
      },
    });
  }
}

export const channelEvents = new ChannelEventBus();