agenthub/tests/push-channel.test.ts

import { afterEach, describe, expect, it, vi } from 'vitest';
import type { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import { bindPushChannel } from '../src/mcp/pushChannel.js';

function sseFrame(seq: number): Uint8Array {
  return new TextEncoder().encode(
    `id: ${seq}\ndata: ${JSON.stringify({
      type: 'task',
      action: 'updated',
      id: 'TSK-0042',
      status: 'open',
      assignedTo: 'codex',
    })}\n\n`,
  );
}

describe('MCP push channel replay cursor', () => {
  const realFetch = globalThis.fetch;

  afterEach(() => {
    globalThis.fetch = realFetch;
    vi.restoreAllMocks();
  });

  it('reconnects with Last-Event-ID after receiving a durable event', async () => {
    const eventHeaders: Headers[] = [];
    let eventRequests = 0;
    let resolveSecondRequest!: () => void;
    const secondRequest = new Promise<void>((resolve) => {
      resolveSecondRequest = resolve;
    });

    globalThis.fetch = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
      const url = String(input);
      if (url.endsWith('/agents/codex/identity')) {
        return new Response(JSON.stringify({ canonical: 'codex', names: ['codex'] }), {
          status: 200,
          headers: { 'Content-Type': 'application/json' },
        });
      }

      eventHeaders.push(new Headers(init?.headers));
      eventRequests += 1;
      if (eventRequests === 1) {
        return new Response(new ReadableStream<Uint8Array>({
          start(controller) {
            controller.enqueue(sseFrame(42));
            controller.close();
          },
        }), { status: 200, headers: { 'Content-Type': 'text/event-stream' } });
      }

      resolveSecondRequest();
      return new Response(new ReadableStream<Uint8Array>({
        start() {
          // Keep the retry connected; the test only needs its request headers.
        },
      }), { status: 200, headers: { 'Content-Type': 'text/event-stream' } });
    }) as typeof fetch;

    const sendLoggingMessage = vi.fn(async () => undefined);
    const fakeServer = {
      server: {
        getClientCapabilities: () => null,
        sendLoggingMessage,
      },
    } as unknown as McpServer;

    bindPushChannel(fakeServer, 'http://agenthub.test', 'codex');
    await secondRequest;

    expect(sendLoggingMessage).toHaveBeenCalledTimes(1);
    expect(eventHeaders).toHaveLength(2);
    expect(eventHeaders[0].get('Last-Event-ID')).toBeNull();
    expect(eventHeaders[1].get('Last-Event-ID')).toBe('42');
  }, 5000);
});