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';
import { startEntityWatcher } from '../src/server/fsWatch.js';
import { createTask, claimTask } from '../src/core/services/taskService.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 (AgentHub-branded)', () => {
  it('every line starts with the AgentHub: prefix', () => {
    const ev: AgentHubEvent = { type: 'decision', action: 'created', id: 'DEC-0001' };
    expect(formatEvent(ev).startsWith('AgentHub: ')).toBe(true);
  });

  it('formats a created task as "Task received <id>  <title>  [status]"', () => {
    const ev: AgentHubEvent = { type: 'task', action: 'created', id: 'TSK-0019', title: 'Magic-App Audit', status: 'open' };
    expect(formatEvent(ev)).toMatch(/^AgentHub: Task received\s+TSK-0019\s+Magic-App Audit\s+\[open\]$/);
  });

  it('formats a created task with no title as "Task received <id>  [status]"', () => {
    const ev: AgentHubEvent = { type: 'task', action: 'created', id: 'TSK-0001', status: 'open' };
    expect(formatEvent(ev)).toMatch(/^AgentHub: Task received\s+TSK-0001\s+\[open\]$/);
  });

  it('formats a completed task as "Task done <id>  by <agent>"', () => {
    const ev: AgentHubEvent = {
      type: 'task',
      action: 'updated',
      id: 'TSK-0003',
      title: 'Implement auth',
      status: 'done',
      role: 'implementer',
      assignedTo: 'windows-claude',
    };
    expect(formatEvent(ev)).toMatch(/^AgentHub: Task done\s+TSK-0003\s+by windows-claude$/);
  });

  it('formats a claimed task as "Task claimed <id>  by <agent>"', () => {
    const ev: AgentHubEvent = { type: 'task', action: 'updated', id: 'TSK-0014', status: 'in_progress', assignedTo: 'codex' };
    expect(formatEvent(ev)).toMatch(/^AgentHub: Task claimed\s+TSK-0014\s+by codex$/);
  });

  it('formats a handoff as "Handoff <id>  → <role>"', () => {
    const ev: AgentHubEvent = { type: 'handoff', action: 'created', id: 'HOF-0012', title: 'For codex', role: 'implementer' };
    expect(formatEvent(ev)).toMatch(/^AgentHub: Handoff\s+HOF-0012\s+.*→ implementer$/);
  });

  it('formats a decision with no detail as just the prefix + id', () => {
    const ev: AgentHubEvent = { type: 'decision', action: 'created', id: 'DEC-0001' };
    expect(formatEvent(ev)).toMatch(/^AgentHub: Decision\s+DEC-0001$/);
  });
});

// ─── 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);
});

// ─── 5. fsWatch: CLI / direct file writes also emit (TSK-0006 part 2) ─────────
// The whole point of the watcher: a `agenthub task create` (no --server) writes
// the entity file directly, bypassing the REST emit path — yet a connected
// watcher must still be triggered. Here we drive the services directly (exactly
// what the CLI does) and assert the eventBus fires.

async function waitFor(
  collected: AgentHubEvent[],
  pred: (e: AgentHubEvent) => boolean,
  ms = 2000,
): Promise<AgentHubEvent | undefined> {
  const deadline = Date.now() + ms;
  while (Date.now() < deadline) {
    const hit = collected.find(pred);
    if (hit) return hit;
    await new Promise((r) => setTimeout(r, 25));
  }
  return collected.find(pred);
}

describe('fsWatch emits for non-REST (CLI/file) writes', () => {
  let cwd: string;
  let stop: () => void;
  const collected: AgentHubEvent[] = [];
  const onChange = (e: AgentHubEvent) => collected.push(e);

  beforeEach(async () => {
    cwd = mkdtempSync(join(tmpdir(), 'ah-fswatch-'));
    init(cwd, { projectName: 'fswatch-test', yes: true });
    collected.length = 0;
    eventBus.on('change', onChange);
    stop = startEntityWatcher(cwd);
    // fs.watch (FSEvents on macOS) needs a brief moment after watch() before it
    // reliably delivers; a write fired in that gap is missed. The server starts
    // its watcher long before any write, so this only matters for the test.
    await new Promise((r) => setTimeout(r, 250));
  });

  afterEach(() => {
    stop();
    eventBus.off('change', onChange);
    collected.length = 0;
    rmSync(cwd, { recursive: true, force: true });
  });

  it('emits task/created when a task is written via the service (CLI path)', async () => {
    const task = createTask(cwd, { title: 'CLI-created task', role: 'implementer' });
    const ev = await waitFor(collected, (e) => e.type === 'task' && e.id === task.id);
    expect(ev).toBeDefined();
    expect(ev).toMatchObject({
      type: 'task',
      action: 'created',
      id: task.id,
      title: 'CLI-created task',
      role: 'implementer',
    });
  }, 4000);

  it('emits task/updated when a task file is mutated via the service (claim)', async () => {
    const task = createTask(cwd, { title: 'T', role: 'implementer' });
    await waitFor(collected, (e) => e.id === task.id); // drain the created event
    collected.length = 0;

    claimTask(cwd, task.id, 'windows-claude');
    const ev = await waitFor(collected, (e) => e.id === task.id && e.action === 'updated');
    expect(ev).toBeDefined();
    expect(ev).toMatchObject({ type: 'task', action: 'updated', status: 'in_progress', assignedTo: 'windows-claude' });
  }, 4000);
});

// ─── 6. fsWatch + REST dedup: each change is delivered exactly once ────────────

describe('fsWatch dedup with the REST emit path', () => {
  let cwd: string;
  let server: Awaited<ReturnType<typeof startServer>>;
  const collected: AgentHubEvent[] = [];
  const onChange = (e: AgentHubEvent) => collected.push(e);

  beforeEach(async () => {
    cwd = mkdtempSync(join(tmpdir(), 'ah-fswatch-dedup-'));
    init(cwd, { projectName: 'fswatch-dedup', yes: true });
    collected.length = 0;
    eventBus.on('change', onChange);
    server = await startServer(cwd, { host: '127.0.0.1', port: 0 });
  });

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

  it('REST POST /tasks emits exactly once (watcher echo suppressed)', async () => {
    await fetch(`${server.url}/tasks`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ title: 'dedup task', role: 'implementer' }),
    });

    // Wait well past the watcher debounce so any duplicate would have landed.
    await new Promise((r) => setTimeout(r, 300));

    const forTask = collected.filter((e) => e.type === 'task' && e.id === 'TSK-0001');
    expect(forTask).toHaveLength(1);
    expect(forTask[0].action).toBe('created');
  }, 4000);
});