agenthub/tests/sse.test.ts

/**
 * Tests for the SSE realtime channel (TSK-0006 part 1).
 *
 * Three concerns:
 *   1. Unit: parseSSEBuffer + formatEvent (pure functions, no I/O).
 *   2. Integration: eventBus emits on REST mutations (via app.inject).
 *   3. E2E: a real server delivers a task/created event over the SSE stream.
 */

import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import { mkdtempSync, rmSync } from 'fs';
import { tmpdir } from 'os';
import { join } from 'path';
import { buildApp } from '../src/server/index.js';
import { startServer } from '../src/server/index.js';
import { eventBus } from '../src/server/events.js';
import type { AgentHubEvent } from '../src/server/events.js';
import { parseSSEBuffer, formatEvent } from '../src/cli/commands/watch.js';
import { init } from '../src/cli/commands/init.js';

// ─── 1. Unit: SSE buffer parser ──────────────────────────────────────────────

describe('parseSSEBuffer', () => {
  it('parses a single complete event', () => {
    const buf = 'data: {"type":"task","action":"created","id":"TSK-0001"}\n\n';
    const { events, remaining } = parseSSEBuffer(buf);
    expect(events).toHaveLength(1);
    expect(events[0].type).toBe('task');
    expect(events[0].action).toBe('created');
    expect(events[0].id).toBe('TSK-0001');
    expect(remaining).toBe('');
  });

  it('parses multiple complete events in one buffer', () => {
    const buf =
      'data: {"type":"task","action":"created","id":"TSK-0001"}\n\n' +
      'data: {"type":"memory","action":"created","id":"MEM-001"}\n\n';
    const { events, remaining } = parseSSEBuffer(buf);
    expect(events).toHaveLength(2);
    expect(events[0].type).toBe('task');
    expect(events[1].type).toBe('memory');
    expect(remaining).toBe('');
  });

  it('ignores SSE keepalive comment blocks', () => {
    const buf =
      ':\n\n' +
      'data: {"type":"handoff","action":"created","id":"HOF-0001"}\n\n';
    const { events } = parseSSEBuffer(buf);
    expect(events).toHaveLength(1);
    expect(events[0].id).toBe('HOF-0001');
  });

  it('keeps an incomplete tail in `remaining`', () => {
    const buf =
      'data: {"type":"task","action":"created","id":"TSK-0001"}\n\n' +
      'data: {"type":"deci'; // truncated chunk
    const { events, remaining } = parseSSEBuffer(buf);
    expect(events).toHaveLength(1);
    expect(remaining).toBe('data: {"type":"deci');
  });

  it('returns empty events and the full buffer when there is no complete event', () => {
    const buf = 'data: {"type":"task"'; // no \n\n yet
    const { events, remaining } = parseSSEBuffer(buf);
    expect(events).toHaveLength(0);
    expect(remaining).toBe('data: {"type":"task"');
  });

  it('silently skips malformed JSON without throwing', () => {
    const buf = 'data: NOT_JSON\n\n';
    const { events } = parseSSEBuffer(buf);
    expect(events).toHaveLength(0);
  });
});

// ─── 2. Unit: formatEvent ────────────────────────────────────────────────────

describe('formatEvent', () => {
  it('formats a minimal event (no optional fields)', () => {
    const ev: AgentHubEvent = { type: 'decision', action: 'created', id: 'DEC-0001' };
    expect(formatEvent(ev)).toBe('[decision/created] DEC-0001');
  });

  it('formats a full task event matching the spec example', () => {
    const ev: AgentHubEvent = {
      type: 'task',
      action: 'updated',
      id: 'TSK-0003',
      title: 'Implement auth',
      status: 'done',
      role: 'implementer',
      assignedTo: 'windows-claude',
    };
    expect(formatEvent(ev)).toBe('[task/updated] TSK-0003 Implement auth status=done role=implementer (windows-claude)');
  });

  it('omits absent optional fields', () => {
    const ev: AgentHubEvent = { type: 'task', action: 'created', id: 'TSK-0001', status: 'open' };
    expect(formatEvent(ev)).toBe('[task/created] TSK-0001 status=open');
  });
});

// ─── 3. Integration: eventBus fires on REST mutations ────────────────────────

describe('eventBus mutations', () => {
  let cwd: string;
  let app: ReturnType<typeof buildApp>;
  const collected: AgentHubEvent[] = [];

  beforeEach(() => {
    cwd = mkdtempSync(join(tmpdir(), 'ah-sse-bus-'));
    init(cwd, { projectName: 'sse-bus-test', yes: true });
    app = buildApp(cwd);
    eventBus.on('change', (e: AgentHubEvent) => collected.push(e));
  });

  afterEach(async () => {
    eventBus.removeAllListeners('change');
    collected.length = 0;
    await app.close();
    rmSync(cwd, { recursive: true, force: true });
  });

  it('emits task/created when POST /tasks succeeds', async () => {
    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'Bus task', role: 'implementer' } });
    expect(collected).toHaveLength(1);
    expect(collected[0]).toMatchObject({ type: 'task', action: 'created', id: 'TSK-0001', title: 'Bus task', role: 'implementer' });
  });

  it('emits task/updated when PATCH /tasks/:id changes status', async () => {
    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'T', role: 'implementer' } });
    collected.length = 0; // clear the created event

    await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'done' } });
    expect(collected).toHaveLength(1);
    expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', id: 'TSK-0001', status: 'done' });
  });

  it('emits task/updated with assignedTo when a task is claimed', async () => {
    await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'T', role: 'implementer' } });
    collected.length = 0;

    await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'in_progress', assignedTo: 'mac-claude' } });
    expect(collected).toHaveLength(1);
    expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', status: 'in_progress', assignedTo: 'mac-claude' });
  });

  it('emits handoff/created when POST /handoffs succeeds', async () => {
    await app.inject({
      method: 'POST',
      url: '/handoffs',
      payload: { fromRole: 'architect', toRole: 'implementer', summary: 'Design complete', context: 'See decisions' },
    });
    expect(collected).toHaveLength(1);
    expect(collected[0]).toMatchObject({ type: 'handoff', action: 'created' });
  });

  it('emits decision/created when POST /decisions succeeds', async () => {
    await app.inject({
      method: 'POST',
      url: '/decisions',
      payload: { title: 'Use SQLite', decision: 'SQLite for local storage', context: 'Low overhead' },
    });
    expect(collected).toHaveLength(1);
    expect(collected[0]).toMatchObject({ type: 'decision', action: 'created', title: 'Use SQLite' });
  });

  it('emits memory/created when POST /memory succeeds', async () => {
    await app.inject({
      method: 'POST',
      url: '/memory',
      payload: { title: 'DNS cache fact', category: 'technical', content: 'TTL matters' },
    });
    expect(collected).toHaveLength(1);
    expect(collected[0]).toMatchObject({ type: 'memory', action: 'created', title: 'DNS cache fact' });
  });

  it('does NOT emit when task creation fails (400)', async () => {
    // title: '' fails TaskSchema min(1) — createTask throws, route returns 400.
    const res = await app.inject({ method: 'POST', url: '/tasks', payload: { title: '' } });
    expect(res.statusCode).toBe(400);
    // No event should be emitted on failure.
    expect(collected).toHaveLength(0);
  });
});

// ─── 4. E2E: SSE stream delivers events over a real HTTP connection ───────────

describe('SSE stream e2e', () => {
  let cwd: string;
  let server: Awaited<ReturnType<typeof startServer>>;

  beforeEach(async () => {
    cwd = mkdtempSync(join(tmpdir(), 'ah-sse-e2e-'));
    init(cwd, { projectName: 'sse-e2e-test', yes: true });
    server = await startServer(cwd, { host: '127.0.0.1', port: 0 });
  });

  afterEach(async () => {
    await server.app.close();
    rmSync(cwd, { recursive: true, force: true });
  });

  it('delivers task/created to a connected SSE subscriber', async () => {
    const controller = new AbortController();

    // Start listening BEFORE posting the task.
    const firstEventPromise = new Promise<AgentHubEvent>((resolve, reject) => {
      const timeout = setTimeout(
        () => reject(new Error('SSE: no event received within 3 s')),
        3000,
      );

      fetch(`${server.url}/events`, { signal: controller.signal })
        .then(async (res) => {
          if (!res.body) {
            clearTimeout(timeout);
            reject(new Error('SSE response has no body'));
            return;
          }
          const reader = res.body.getReader();
          const dec = new TextDecoder();
          let buf = '';

          while (true) {
            let done: boolean;
            let value: Uint8Array | undefined;
            try {
              ({ done, value } = await reader.read());
            } catch {
              break; // AbortError when controller.abort() is called
            }
            if (done) break;
            if (value) buf += dec.decode(value, { stream: true });

            const { events, remaining } = parseSSEBuffer(buf);
            buf = remaining;

            for (const ev of events) {
              clearTimeout(timeout);
              resolve(ev);
              return;
            }
          }
        })
        .catch((err: unknown) => {
          // AbortError is expected when we abort after receiving the event.
          if (err instanceof Error && err.name === 'AbortError') return;
          reject(err);
        });
    });

    // Give the SSE connection a moment to be established before posting.
    await new Promise((r) => setTimeout(r, 80));

    // Trigger a mutation through the REST API.
    await fetch(`${server.url}/tasks`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ title: 'SSE e2e task', role: 'implementer' }),
    });

    const event = await firstEventPromise;
    controller.abort(); // stop the SSE listener

    expect(event.type).toBe('task');
    expect(event.action).toBe('created');
    expect(event.id).toBe('TSK-0001');
    expect(event.title).toBe('SSE e2e task');
    expect(event.role).toBe('implementer');
  }, 5000);

  it('role filter on /events?role= suppresses non-matching tasks', async () => {
    const controller = new AbortController();
    const received: AgentHubEvent[] = [];

    // Subscribe with role=architect — should NOT see the implementer task.
    const listenDone = fetch(`${server.url}/events?role=architect`, { signal: controller.signal })
      .then(async (res) => {
        if (!res.body) return;
        const reader = res.body.getReader();
        const dec = new TextDecoder();
        let buf = '';
        while (true) {
          let done: boolean;
          let value: Uint8Array | undefined;
          try {
            ({ done, value } = await reader.read());
          } catch {
            break;
          }
          if (done) break;
          if (value) buf += dec.decode(value, { stream: true });
          const { events, remaining } = parseSSEBuffer(buf);
          buf = remaining;
          received.push(...events);
        }
      })
      .catch((err: unknown) => {
        if (err instanceof Error && err.name === 'AbortError') return;
        throw err;
      });

    await new Promise((r) => setTimeout(r, 80));

    // Post a task for 'implementer' role — should be filtered out.
    await fetch(`${server.url}/tasks`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ title: 'Implementer task', role: 'implementer' }),
    });

    // Give time for any event to arrive (it shouldn't).
    await new Promise((r) => setTimeout(r, 150));

    controller.abort();
    await listenDone;

    expect(received).toHaveLength(0);
  }, 5000);
});